Spring 整合 WebSocket 實戰（下）：消息推送服務、定時任務與集群擴展

核心業務：消息推送服務實現

@Component
public class MessagePushService implements ApplicationListener<ApplicationEvent> {

    private static final Logger log = LoggerFactory.getLogger(MessagePushService.class);
    private final Gson gson = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create();
    private final ExecutorService executor = Executors.newCachedThreadPool();
    @Autowired private NotificationService notificationService;

    /** 用戶上線：推送未讀消息摘要 */
    public void onUserConnected(String userKey) {
        executor.execute(() -> {
            try {
                List<NotificationSummary> summaries = notificationService.getUnreadSummaries(userKey);
                if (!summaries.isEmpty()) {
                    pushToUser(userKey, MessageType.UNREAD_SUMMARY, summaries);
                }
            } catch (Exception e) {
                log.error("上線推送失敗: {}", userKey, e);
            }
        });
    }

    /** 業務事件觸發：即時推送 */
    public void pushNotification(NotificationEvent event) {
        executor.execute(() -> {
            String targetUser = event.getTargetUser();
            UserSessionVo vo = SessionRegistry.get(targetUser);
            if (vo != null && vo.getSession() != null && vo.getSession().isOpen()) {
                try {
                    vo.getSession().sendMessage(new TextMessage(gson.toJson(event)));
                    vo.setLastPushTime(new Date());
                } catch (IOException e) {
                    log.error("推送失敗: {}", targetUser, e);
                    // 可選：存入離線表、發送郵件/短信補償
                }
            } else {
                // 用戶離線：持久化離線消息
                notificationService.saveOffline(event);
            }
        });
    }

    /** 定時推送：如每日早報、待辦提醒 */
    public void scheduleDailyDigest(String userKey, LocalTime time) {
        // 使用 Quartz 或 Spring @Scheduled(cron = "0 0 8 * * ?")
        // 此處示例：Spring TaskScheduler
    }

    private void pushToUser(String userKey, MessageType type, Object payload) {
        UserSessionVo vo = SessionRegistry.get(userKey);
        if (vo != null && vo.getSession() != null && vo.getSession().isOpen()) {
            try {
                Map<String, Object> msg = new HashMap<>();
                msg.put("type", type.name());
                msg.put("data", payload);
                msg.put("timestamp", System.currentTimeMillis());
                vo.getSession().sendMessage(new TextMessage(gson.toJson(msg)));
            } catch (IOException e) {
                log.error("定向推送失敗: {}", userKey, e);
            }
        }
    }
}

定時任務整合：Quartz 實現精準推送

@Component
public class ScheduledPushJob implements Job {

    @Autowired private MessagePushService pushService;
    @Autowired private NotificationService notificationService;

    @Override
    public void execute(JobExecutionContext context) {
        String userKey = context.getMergedJobDataMap().getString("userKey");
        Long notificationId = context.getMergedJobDataMap().getLong("notificationId");

        Notification n = notificationService.getById(notificationId);
        if (n != null) {
            pushService.pushToUser(userKey, MessageType.SCHEDULED, n);
        }
    }
}

// 動態創建定時任務
public void createTimingPush(Notification n) {
    JobDetail job = JobBuilder.newJob(ScheduledPushJob.class)
        .withIdentity("push_" + n.getId(), "notifications")
        .usingJobData("userKey", n.getTargetUser())
        .usingJobData("notificationId", n.getId())
        .build();

    Trigger trigger = TriggerBuilder.newTrigger()
        .withIdentity("trigger_" + n.getId(), "notifications")
        .startAt(Date.from(n.getPushTime().atZone(ZoneId.systemDefault()).toInstant()))
        .withSchedule(SimpleScheduleBuilder.simpleSchedule().withRepeatCount(0))
        .build();

    try {
        scheduler.scheduleJob(job, trigger);
        if (!scheduler.isStarted()) scheduler.start();
    } catch (SchedulerException e) {
        throw new RuntimeException("定時任務創建失敗", e);
    }
}

集群環境擴展方案

問題

• 本地 ConcurrentHashMap 無法跨節點共享會話
• 消息發送到用戶所在節點需路由機制

解決架構

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  Nginx      │────▶│  App Node 1 │     │  App Node 2 │
│  (Sticky    │     │  Session    │     │  Session    │
│   Session)  │     │  Registry   │     │  Registry   │
└─────────────┘     └──────┬──────┘     └──────┬──────┘
                           │                   │
                    ┌──────▼───────────────────▼──────┐
                    │     Redis Cluster               │
                    │  - Session Registry (Hash)      │
                    │  - Pub/Sub Channel (Message Bus)│
                    └─────────────────────────────────┘

關鍵組件改造

1. SessionRegistry：改用 RedisTemplate.opsForHash() 存儲 userKey → SessionInfo(JSON)
2. 消息路由：發送方發布到 Redis Channel ws:push:{userKey}，各節點訂閱並檢查本地是否持有該會話，有則發送
3. 心跳檢測：定時掃描 Redis 中會話最後活躍時間，清理僵屍連接

Spring 配置啟用

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableSimpleBroker("/topic", "/queue"); // 簡易代理
        // 生產建議：registry.enableStompBrokerRelay("/topic", "/queue")
        //          .setRelayHost("rabbitmq-host").setRelayPort(61613);
        registry.setApplicationDestinationPrefixes("/app");
    }
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws").setAllowedOrigins("*").withSockJS();
    }
}

使用 STOMP 子協議 + SockJS 降級，前端引入 stomp.js + sockjs.min.js 即可統一收發。