WebSocket 深度

概述

Spring MVC 对 WebSocket 的支持建立在 JSR-356 之上，spring-websocket 模块提供了从 HTTP 握手升级到 WebSocket 连接的完整支持。核心能力包括：低级 WebSocketHandler API、STOMP 协议支持、@MessageMapping 注解处理、用户队列、广播与点对点通信，以及通过 Redis Pub/Sub 实现的集群广播。

1. WebSocketHandler 接口源码

WebSocketHandler 是 Spring WebSocket 最底层的核心接口，定义了会话生命周期方法。

1.1 接口定义

java

// org.springframework.web.socket.WebSocketHandler
public interface WebSocketHandler {
    void afterConnectionEstablished(WebSocketSession session) throws Exception;
    void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception;
    void handleTransportError(WebSocketSession session, Throwable exception) throws Exception;
    void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception;
    boolean supportsPartialMessages();
}

1.2 AbstractWebSocketHandler

Spring 提供了 TextWebSocketHandler 和 BinaryWebSocketHandler 两个抽象类，按消息类型分派：

java

public abstract class AbstractWebSocketHandler implements WebSocketHandler {
    @Override
    public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
        if (message instanceof TextMessage) handleTextMessage(session, (TextMessage) message);
        else if (message instanceof BinaryMessage) handleBinaryMessage(session, (BinaryMessage) message);
        else if (message instanceof PongMessage) handlePongMessage(session, (PongMessage) message);
    }
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {}
    protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception {}
    protected void handlePongMessage(WebSocketSession session, PongMessage message) throws Exception {}
    // 其他方法空实现, supportsPartialMessages() 返回 false
}

1.3 自定义 Handler

java

public class ChatHandler extends TextWebSocketHandler {
    private static final Map<String, WebSocketSession> sessions = new ConcurrentHashMap<>();

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        String userId = (String) session.getAttributes().get("userId");
        sessions.put(userId, session);
    }

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        session.sendMessage(new TextMessage("回复: " + message.getPayload()));
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        sessions.remove(session.getAttributes().get("userId"));
    }
}

1.4 WebSocketSession 核心方法

java

public interface WebSocketSession {
    String getId();                          // 会话唯一 ID
    URI getUri();                            // 连接 URI
    Map<String, Object> getAttributes();     // 会话属性（来自 HTTP 握手）
    Principal getPrincipal();                // 当前认证用户
    InetSocketAddress getLocalAddress();     // 本地地址
    InetSocketAddress getRemoteAddress();    // 远程地址
    boolean isOpen();                        // 连接是否仍在打开状态
    void sendMessage(WebSocketMessage<?> message) throws IOException;
    void close(CloseStatus closeStatus) throws IOException;
}

1.5 配置 WebSocket 端点

java

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(new ChatHandler(), "/chat")
                .setAllowedOrigins("*").withSockJS();
    }
}

2. STOMP 协议详解

2.1 什么是 STOMP

STOMP（Simple/Streaming Text Oriented Messaging Protocol）是基于 WebSocket 的文本消息协议，提供类似于 JMS 的发布/订阅模型，包含帧（Frame）、目的地（Destination）、订阅（Subscription）和发布（Publish）四个核心概念。

2.2 帧格式

STOMP 帧由命令、头部和体组成，以 NULL 字符（\0）结束：

text

COMMAND
header1:value1
header2:value2

Body content^@

客户端帧：

命令	说明	示例
`CONNECT`	建立连接	`CONNECT\naccept-version:1.2\nhost:localhost\n\n\0`
`SUBSCRIBE`	订阅目的地	`SUBSCRIBE\nid:sub-1\ndestination:/topic/news\n\n\0`
`SEND`	发送消息	`SEND\ndestination:/app/chat\n\nHello\0`
`DISCONNECT`	断开连接	`DISCONNECT\nreceipt:r1\n\n\0`

服务端帧：

命令	说明	示例
`CONNECTED`	确认连接	`CONNECTED\nversion:1.2\nheart-beat:10000,10000\n\n\0`
`MESSAGE`	推送消息	`MESSAGE\nsubscription:sub-1\ndestination:/topic/news\n\n正文\0`
`RECEIPT`	回执确认	`RECEIPT\nreceipt-id:r1\n\n\0`
`ERROR`	错误信息	`ERROR\nmessage:Invalid frame\n\n详情\0`

2.3 订阅/发布模型

前缀	用途	说明
`/topic`	广播	所有订阅者接收
`/queue`	点对点	消息投递给一个消费者
`/user`	用户队列	消息投递给特定用户

2.4 握手流程

text

客户端                         服务端
  │  1. HTTP Upgrade 请求       │
  │ ──────────────────────────> │
  │  101 Switching Protocols    │
  │ <────────────────────────── │
  │  2. STOMP CONNECT           │
  │ ──────────────────────────> │
  │  CONNECTED                  │
  │ <────────────────────────── │
  │  3. SUBSCRIBE /topic/news   │
  │ ──────────────────────────> │
  │  4. SEND /app/message       │
  │ ──────────────────────────> │
  │  MESSAGE /topic/news        │
  │ <────────────────────────── │

2.5 Spring 配置 STOMP

java

@Configuration
@EnableWebSocketMessageBroker
public class StompConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableSimpleBroker("/topic", "/queue");           // 广播代理
        registry.setApplicationDestinationPrefixes("/app");         // 客户端发送前缀
        registry.setUserDestinationPrefix("/user");                 // 用户队列前缀
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws-stomp").setAllowedOriginPatterns("*").withSockJS();
    }
}

3. @MessageMapping 注解处理

@MessageMapping 是 Spring WebSocket 消息处理的核心注解，类似 HTTP 的 @RequestMapping。

3.1 基本用法

java

@Controller
public class ChatController {

    @MessageMapping("/chat.sendMessage")
    @SendTo("/topic/public")
    public ChatMessage sendMessage(@Payload ChatMessage msg, Principal principal) {
        msg.setSender(principal.getName());
        msg.setTimestamp(LocalDateTime.now());
        return msg;
    }

    @MessageMapping("/chat.addUser")
    @SendTo("/topic/public")
    public ChatMessage addUser(@Payload ChatMessage msg, SimpMessageHeaderAccessor headerAccessor) {
        headerAccessor.getSessionAttributes().put("username", msg.getSender());
        return msg;
    }
}

3.2 支持的参数类型

参数类型	说明
`@Payload`	消息体（JSON 自动反序列化）
`@Header`	消息头中的指定字段
`@Headers`	所有消息头
`@DestinationVariable`	路径变量
`Principal`	当前认证用户
`SimpMessageHeaderAccessor`	消息头访问器

3.3 路径映射与通配符

java

@Controller
public class RoomController {

    // 路径模板：/app/room/{roomId}/join
    @MessageMapping("/room/{roomId}/join")
    @SendTo("/topic/room/{roomId}")
    public JoinMessage joinRoom(@DestinationVariable String roomId, @Payload JoinMessage msg) {
        msg.setRoomId(roomId);
        return msg;
    }

    // 通配符：/app/chat/**
    @MessageMapping("/chat/**")
    public void handleWildcard(@Payload String message) { }
}

3.4 异常处理

java

@Controller
public class ExceptionController {

    @MessageMapping("/risky")
    public String risky(String input) {
        if (input == null || input.isBlank())
            throw new IllegalArgumentException("输入不能为空");
        return "成功";
    }

    @MessageExceptionHandler
    @SendToUser("/queue/errors")
    public String handleException(IllegalArgumentException e) {
        return "错误: " + e.getMessage();
    }
}

4. 用户队列（@SendToUser / UserDestinationResolver）

4.1 @SendToUser 基础

@SendToUser 将消息发送给当前认证用户，实现点对点通信：

java

@Controller
public class UserController {

    // 回复发送者：/user/{username}/queue/private
    @MessageMapping("/private/message")
    @SendToUser("/queue/private")
    public PrivateMessage handlePrivate(@Payload PrivateMessage msg, Principal principal) {
        msg.setFrom(principal.getName());
        return msg;
    }

    // broadcast=true 广播给所有用户（含发送者）
    @MessageMapping("/broadcast")
    @SendToUser(value = "/queue/broadcast", broadcast = true)
    public String broadcastToAll(String msg) { return "广播: " + msg; }
}

4.2 底层原理：UserDestinationResolver

java

public interface UserDestinationResolver {
    UserDestinationResult resolveDestination(Message<?> message);
}

// 解析结果包含：源路径 /user/queue/private
// 用户路径 /queue/private-user123
// 当前用户名等

处理流程：@SendToUser → UserDestinationResolver 从 Principal 解析用户名 → 转换路径 /user/queue/private → /queue/private-user123 → 匹配会话并发送。

4.3 手动发送（SimpMessagingTemplate）

java

@Service
public class NotificationService {

    private final SimpMessagingTemplate messagingTemplate;

    public NotificationService(SimpMessagingTemplate messagingTemplate) {
        this.messagingTemplate = messagingTemplate;
    }

    public void sendToUser(String username, String destination, Object payload) {
        messagingTemplate.convertAndSendToUser(username, destination, payload);
    }

    public void broadcast(String destination, Object payload) {
        messagingTemplate.convertAndSend(destination, payload);
    }
}

5. 广播与点对点通信

5.1 广播实现

java

@Controller
public class BroadcastController {

    @MessageMapping("/system/announcement")
    @SendTo("/topic/announcement")
    public SystemAnnouncement broadcast(@Payload AnnouncementRequest req) {
        return new SystemAnnouncement(req.getTitle(), req.getContent(), LocalDateTime.now(), "SYSTEM");
    }
}

JS 客户端订阅广播：

javascript

const client = new StompJs.Client({ brokerURL: 'ws://localhost:8080/ws-stomp' });
client.onConnect = () => {
    client.subscribe('/topic/announcement', msg => {
        const ann = JSON.parse(msg.body);
        console.log('公告:', ann);
    });
    client.subscribe('/topic/public', msg => {
        appendChat(JSON.parse(msg.body));
    });
};
client.activate();

5.2 点对点通信

java

@Controller
public class PrivateController {

    private final SimpMessagingTemplate template;

    public PrivateController(SimpMessagingTemplate template) { this.template = template; }

    @MessageMapping("/private/send")
    public void sendPrivate(@Payload PrivateRequest req, Principal sender) {
        ChatMessage msg = new ChatMessage();
        msg.setFrom(sender.getName());
        msg.setTo(req.getTargetUser());
        msg.setContent(req.getContent());
        msg.setTimestamp(System.currentTimeMillis());
        template.convertAndSendToUser(req.getTargetUser(), "/queue/private", msg);
    }
}

对比项	广播	点对点
目的地前缀	`/topic`	`/queue` + 用户解析
接收者	所有订阅者	特定用户
实现	`@SendTo` / `convertAndSend`	`@SendToUser` / `convertAndSendToUser`
场景	公告、全局通知	私信、个人提醒

5.3 在线状态管理

java

@Component
public class SessionManager {
    private final ConcurrentHashMap<String, CopyOnWriteArrayList<String>> userSessions = new ConcurrentHashMap<>();
    private final SimpMessagingTemplate template;

    public SessionManager(SimpMessagingTemplate template) { this.template = template; }

    public void registerSession(String userId, String sessionId) {
        userSessions.computeIfAbsent(userId, k -> new CopyOnWriteArrayList<>()).add(sessionId);
        template.convertAndSend("/topic/online-status", new OnlineStatus(userId, true));
    }

    public void removeSession(String userId, String sessionId) {
        List<String> sessions = userSessions.get(userId);
        if (sessions != null && sessions.remove(sessionId) && sessions.isEmpty()) {
            userSessions.remove(userId);
            template.convertAndSend("/topic/online-status", new OnlineStatus(userId, false));
        }
    }

    public boolean isOnline(String userId) {
        return userSessions.containsKey(userId) && !userSessions.get(userId).isEmpty();
    }
}

6. 集群广播（Redis Pub/Sub + WebSocket 集群）

6.1 问题背景

单机 SimpleBroker 在内存中转发消息，集群中用户 A 在节点 1、用户 B 在节点 2 时跨节点消息无法送达。解决方案：通过 Redis Pub/Sub 实现跨节点消息转发。

6.2 架构

text

                 ┌──────────────────┐
                 │     Redis        │
                 │  Pub/Sub Channel │
                 └──┬────────────┬──┘
                    │            │
              ┌─────▼────┐ ┌─────▼────┐
              │ Node 1   │ │ Node 2   │
              │  Sessions│ │  Sessions│
              │  ┌───┐   │ │  ┌───┐   │
              │  │ A │   │ │  │ B │   │
              │  └───┘   │ │  └───┘   │
              └──────────┘ └──────────┘

6.3 Redis 集群代理实现

java

@Component
public class RedisClusterBroker {

    private static final String CHANNEL = "websocket:cluster";
    private final SimpMessagingTemplate template;
    private final StringRedisTemplate redisTemplate;
    private final ObjectMapper mapper;

    public RedisClusterBroker(SimpMessagingTemplate t, StringRedisTemplate r, ObjectMapper m) {
        this.template = t; this.redisTemplate = r; this.mapper = m;
    }

    public void sendToUserInCluster(String userId, String dest, Object payload) {
        // 先本地发送
        template.convertAndSendToUser(userId, dest, payload);
        // 再广播到集群
        publish(new ClusterMessage("USER", userId, dest, payload));
    }

    public void broadcastToCluster(String dest, Object payload) {
        publish(new ClusterMessage("BROADCAST", null, dest, payload));
    }

    private void publish(ClusterMessage msg) {
        try { redisTemplate.convertAndSend(CHANNEL, mapper.writeValueAsString(msg)); }
        catch (Exception e) { System.err.println("集群消息发布失败: " + e.getMessage()); }
    }

    @Component
    public static class Subscriber {
        private final SimpMessagingTemplate template;
        private final ObjectMapper mapper;
        public Subscriber(SimpMessagingTemplate t, ObjectMapper m) { this.template = t; this.mapper = m; }

        @EventListener
        public void handle(Message message) {
            try {
                ClusterMessage msg = mapper.readValue(message.getBody(), ClusterMessage.class);
                if ("USER".equals(msg.type()))
                    template.convertAndSendToUser(msg.userId(), msg.destination(), msg.payload());
                else
                    template.convertAndSend(msg.destination(), msg.payload());
            } catch (Exception e) { System.err.println("处理集群消息失败"); }
        }
    }
}

record ClusterMessage(String type, String userId, String destination, Object payload) { }

6.4 Redis Pub/Sub 配置

yaml

spring:
  redis:
    host: localhost
    port: 6379
    timeout: 3000ms
    lettuce:
      pool: { max-active: 8, max-idle: 4, min-idle: 0 }

java

@Configuration
public class RedisPubSubConfig {
    @Bean
    public RedisMessageListenerContainer container(
            RedisConnectionFactory factory, RedisClusterBroker.Subscriber sub) {
        RedisMessageListenerContainer c = new RedisMessageListenerContainer();
        c.setConnectionFactory(factory);
        c.addMessageListener(new MessageListenerAdapter(sub, "handle"), new PatternTopic("websocket:cluster"));
        return c;
    }
}

7. 心跳保活（Heartbeat）

7.1 原理

WebSocket 的 Ping/Pong 帧结合 STOMP 心跳协商，确保连接双方在空闲时感知对方在线。

text

服务端                     客户端
  │ CONNECTED               │
  │ heart-beat:10000,10000  │
  │ <────────────────────── │
  │   (10秒静默)            │
  │ ─── Ping ─────────────> │
  │ <─── Pong ───────────── │

7.2 Spring 心跳配置

java

@Configuration
@EnableWebSocketMessageBroker
public class HeartbeatConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableSimpleBroker("/topic", "/queue")
                .setHeartbeatValue(new long[]{10000, 10000})
                .setTaskScheduler(heartbeatScheduler());
    }

    @Bean
    public TaskScheduler heartbeatScheduler() {
        ThreadPoolTaskScheduler s = new ThreadPoolTaskScheduler();
        s.setPoolSize(2);
        s.setThreadNamePrefix("ws-heartbeat-");
        return s;
    }
}

7.3 客户端心跳（JavaScript）

javascript

class WebSocketClient {
    constructor(url) {
        this.url = url;
        this.ws = null;
        this.heartbeatInterval = null;
        this.reconnectAttempts = 0;
        this.HEARTBEAT = 10000;
        this.MAX_RETRY = 10;
    }

    connect() {
        this.ws = new WebSocket(this.url);
        this.ws.onopen = () => { this.reconnectAttempts = 0; this.startHeartbeat(); };
        this.ws.onmessage = (e) => { if (e.data !== 'pong') this.handleMessage(e.data); };
        this.ws.onclose = () => { this.stopHeartbeat(); this.reconnect(); };
    }

    startHeartbeat() {
        this.heartbeatInterval = setInterval(() => {
            if (this.ws?.readyState === WebSocket.OPEN) this.ws.send('ping');
        }, this.HEARTBEAT);
    }

    stopHeartbeat() { if (this.heartbeatInterval) clearInterval(this.heartbeatInterval); }

    reconnect() {
        if (this.reconnectAttempts >= this.MAX_RETRY) return;
        this.reconnectAttempts++;
        setTimeout(() => this.connect(), 3000);
    }

    send(data) { this.ws?.send(typeof data === 'string' ? data : JSON.stringify(data)); }
    disconnect() { this.stopHeartbeat(); this.ws?.close(); }
}

8. 实战：在线客服系统 + 未读消息 + Redis 集群广播

8.1 系统设计

基于 Spring WebSocket + STOMP + Redis Pub/Sub 构建在线客服系统，核心功能：用户与客服聊天、未读消息计数、集群消息同步、在线状态管理。

架构：

text

用户端 ─┐              ┌─ Nginx 负载均衡
客服端 ─┼──────────────┤
客服端 ─┘              └──┬── Node 1 ───┐
                          └── Node 2 ───┼── Redis Pub/Sub
                                        └── MySQL

8.2 实体模型

java

@Data
public class ChatMessage implements Serializable {
    private String id, senderId, senderName, receiverId, content, conversationId;
    private String type;  // CHAT / JOIN / LEAVE / TYPING / READ / SYSTEM
    private Long timestamp;
    private Boolean read;
}

@Data
public class Conversation implements Serializable {
    private String conversationId, userId, csId, status; // ACTIVE / CLOSED / WAITING
    private Long createdAt, updatedAt;
    private Integer unreadCount;
}

8.3 客服控制器

java

@Controller
public class CustomerServiceController {

    private final SimpMessagingTemplate template;
    private final RedisUnreadCounter unreadCounter;
    private final RedisClusterBroker clusterBroker;

    public CustomerServiceController(SimpMessagingTemplate t, RedisUnreadCounter u, RedisClusterBroker c) {
        this.template = t; this.unreadCounter = u; this.clusterBroker = c;
    }

    @MessageMapping("/cs/connect")
    @SendToUser("/queue/cs/connected")
    public Conversation connect(@Payload ConnectRequest req, Principal p) {
        String userId = p.getName(), csId = "cs-001";  // 简化分配
        Conversation conv = new Conversation();
        conv.setConversationId(UUID.randomUUID().toString()); conv.setUserId(userId);
        conv.setCsId(csId); conv.setStatus("ACTIVE"); conv.setCreatedAt(System.currentTimeMillis());
        conv.setUnreadCount(0);
        clusterBroker.sendToUserInCluster(csId, "/queue/cs/new-user", conv);
        return conv;
    }

    @MessageMapping("/cs/send")
    public void send(@Payload ChatMessage msg, Principal p) {
        msg.setId(UUID.randomUUID().toString()); msg.setSenderId(p.getName());
        msg.setTimestamp(System.currentTimeMillis()); msg.setRead(false);
        unreadCounter.increment(msg.getReceiverId(), msg.getConversationId());
        clusterBroker.sendToUserInCluster(msg.getReceiverId(), "/queue/cs/message", msg);
    }

    @MessageMapping("/cs/read")
    public void markRead(@Payload ReadReceipt receipt, Principal p) {
        unreadCounter.clear(receipt.getConversationId(), p.getName());
        clusterBroker.sendToUserInCluster(receipt.getOtherUserId(), "/queue/cs/read-receipt", receipt);
    }
}

8.4 未读消息管理（Redis Hash）

java

@Component
public class RedisUnreadCounter {
    private static final String KEY_PREFIX = "unread:";
    private final StringRedisTemplate redis;

    public RedisUnreadCounter(StringRedisTemplate redis) { this.redis = redis; }

    public void increment(String userId, String conversationId) {
        redis.opsForHash().increment(KEY_PREFIX + userId, conversationId, 1);
    }

    public Map<String, Integer> getUnreadCounts(String userId) {
        Map<Object, Object> entries = redis.opsForHash().entries(KEY_PREFIX + userId);
        Map<String, Integer> result = new HashMap<>();
        entries.forEach((k, v) -> result.put((String) k, Integer.parseInt((String) v)));
        return result;
    }

    public int getTotalUnread(String userId) {
        return getUnreadCounts(userId).values().stream().mapToInt(Integer::intValue).sum();
    }

    public void clear(String conversationId, String userId) {
        redis.opsForHash().delete(KEY_PREFIX + userId, conversationId);
    }
}

8.5 集群版在线状态管理

java

@Component
public class RedisOnlineStatusManager {
    private static final String ONLINE_KEY = "online:users";
    private static final String SESSION_PREFIX = "user:sessions:";
    private final StringRedisTemplate redis;

    public RedisOnlineStatusManager(StringRedisTemplate redis) { this.redis = redis; }

    public void userOnline(String userId, String sessionId, String nodeId) {
        redis.opsForSet().add(ONLINE_KEY, userId);
        redis.opsForHash().put(SESSION_PREFIX + userId, sessionId, "{\"nodeId\":\"" + nodeId + "\"}");
    }

    public void userOffline(String userId, String sessionId) {
        redis.opsForHash().delete(SESSION_PREFIX + userId, sessionId);
        if (redis.opsForHash().size(SESSION_PREFIX + userId) == 0)
            redis.opsForSet().remove(ONLINE_KEY, userId);
    }

    public boolean isOnline(String userId) {
        return Boolean.TRUE.equals(redis.opsForSet().isMember(ONLINE_KEY, userId));
    }

    public Set<String> getAllOnline() { return redis.opsForSet().members(ONLINE_KEY); }
    public Long getOnlineCount() { return redis.opsForSet().size(ONLINE_KEY); }
}

8.6 WebSocket 配置

yaml

server:
  port: 8080
spring:
  application: { name: cs-websocket-service }
  redis:
    host: ${REDIS_HOST:localhost}
    port: ${REDIS_PORT:6379}
    timeout: 5000ms
    lettuce: { pool: { max-active: 16, max-idle: 8, min-idle: 2 } }

java

@Configuration
@EnableWebSocketMessageBroker
public class CsWebSocketConfig implements WebSocketMessageBrokerConfigurer {
    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableSimpleBroker("/topic", "/queue")
                .setHeartbeatValue(new long[]{10000, 10000})
                .setTaskScheduler(scheduler());
        registry.setApplicationDestinationPrefixes("/app");
        registry.setUserDestinationPrefix("/user");
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws-cs").setAllowedOriginPatterns("*").withSockJS();
    }

    @Bean
    public TaskScheduler scheduler() {
        ThreadPoolTaskScheduler s = new ThreadPoolTaskScheduler();
        s.setPoolSize(4); s.setThreadNamePrefix("ws-hb-"); return s;
    }
}

8.7 前端集成（JavaScript + STOMP）

javascript

class CustomerServiceClient {
    constructor() {
        this.client = null; this.userId = null;
        this.conversations = new Map(); this.unreadCount = 0;
        this.callbacks = [];
    }

    connect(userId) {
        this.userId = userId;
        this.client = new StompJs.Client({
            brokerURL: `ws://${location.host}/ws-cs`,
            reconnectDelay: 5000,
            heartbeatIncoming: 10000,
            heartbeatOutgoing: 10000,
            onConnect: () => this.onConnected(),
            onStompError: (f) => console.error('STOMP错误:', f.headers.message)
        });
        this.client.activate();
    }

    onConnected() {
        this.client.subscribe('/user/queue/cs/message', (m) => this.handleMsg(JSON.parse(m.body)));
        this.client.subscribe('/user/queue/cs/new-user', (m) => this.handleNew(JSON.parse(m.body)));
        this.client.subscribe('/user/queue/cs/read-receipt', (m) => this.handleRead(JSON.parse(m.body)));
        this.client.subscribe('/user/queue/cs/connected', (m) => this.handleConn(JSON.parse(m.body)));
        this.client.publish({ destination: '/app/cs/connect', body: JSON.stringify({ userId: this.userId }) });
    }

    sendMsg(receiverId, content, conversationId) {
        this.client.publish({ destination: '/app/cs/send', body: JSON.stringify({ receiverId, content, conversationId }) });
    }

    markRead(conversationId, otherUserId) {
        this.client.publish({ destination: '/app/cs/read', body: JSON.stringify({ conversationId, otherUserId }) });
    }

    handleMsg(msg) {
        if (!msg.read && msg.senderId !== this.userId) {
            this.unreadCount++;
            document.getElementById('unread-badge').textContent = this.unreadCount > 99 ? '99+' : this.unreadCount;
        }
        this.callbacks.forEach(cb => cb(msg)); this.appendMessage(msg);
    }

    handleNew(conv) { this.conversations.set(conv.conversationId, conv); this.renderList(); }
    handleRead(receipt) { document.querySelectorAll(`[data-msg-id="${receipt.messageId}"]`).forEach(el => el.classList.add('read')); }
    handleConn(conv) { this.conversations.set(conv.conversationId, conv); this.renderList(); }
    onMessage(cb) { this.callbacks.push(cb); }
}

8.8 Maven 依赖

xml

<dependencies>
    <dependency><groupId>org.springframework</groupId><artifactId>spring-websocket</artifactId><version>6.1.12</version></dependency>
    <dependency><groupId>org.springframework</groupId><artifactId>spring-messaging</artifactId><version>6.1.12</version></dependency>
    <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId><version>3.3.3</version></dependency>
    <dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId><version>2.17.2</version></dependency>
    <dependency><groupId>org.webjars</groupId><artifactId>sockjs-client</artifactId><version>1.5.1</version></dependency>
    <dependency><groupId>org.webjars</groupId><artifactId>stomp-websocket</artifactId><version>2.3.4</version></dependency>
</dependencies>

9. 总结

9.1 技术选型

场景	推荐方案
实时通知、简单广播	`SimpleBroker` + `@SendTo`
私信、用户定向通知	`@SendToUser` + `SimpMessagingTemplate`
多节点集群部署	Redis Pub/Sub 或 RabbitMQ
高吞吐量企业级	外部 STOMP Broker（RabbitMQ/ActiveMQ）
移动端弱网场景	SockJS 降级 + 心跳保活

9.2 关键要点

WebSocketHandler：底层接口，适合完全控制连接生命周期
STOMP：发布/订阅模型简化消息路由，推荐大多数场景使用
集群通信：Redis Pub/Sub 轻量级方案适合中小规模；大型系统推荐 RabbitMQ
心跳机制：防止 NAT 网关超时断开，建议 10-30 秒间隔
未读消息：Redis Hash 比数据库更适合高频读写
安全性：生产环境配置 setAllowedOrigins 严格限定来源，集成 Spring Security

9.3 参考资料

Spring WebSocket 文档：https://docs.spring.io/spring-framework/reference/web/websocket.html
STOMP 协议规范：https://stomp.github.io/stomp-specification-1.2.html
SockJS：https://github.com/sockjs/sockjs-client
Redis Pub/Sub：https://redis.io/docs/manual/pubsub/

WebSocket 深度 ​

概述 ​

1. WebSocketHandler 接口源码 ​

1.1 接口定义 ​

1.2 AbstractWebSocketHandler ​

1.3 自定义 Handler ​

1.4 WebSocketSession 核心方法 ​

1.5 配置 WebSocket 端点 ​

2. STOMP 协议详解 ​

2.1 什么是 STOMP ​

2.2 帧格式 ​

2.3 订阅/发布模型 ​

2.4 握手流程 ​

2.5 Spring 配置 STOMP ​

3. @MessageMapping 注解处理 ​

3.1 基本用法 ​

3.2 支持的参数类型 ​

3.3 路径映射与通配符 ​

3.4 异常处理 ​

4. 用户队列（@SendToUser / UserDestinationResolver） ​

4.1 @SendToUser 基础 ​

4.2 底层原理：UserDestinationResolver ​

4.3 手动发送（SimpMessagingTemplate） ​

5. 广播与点对点通信 ​

5.1 广播实现 ​

5.2 点对点通信 ​

5.3 在线状态管理 ​

6. 集群广播（Redis Pub/Sub + WebSocket 集群） ​

6.1 问题背景 ​

6.2 架构 ​

6.3 Redis 集群代理实现 ​

6.4 Redis Pub/Sub 配置 ​

7. 心跳保活（Heartbeat） ​

7.1 原理 ​

7.2 Spring 心跳配置 ​

7.3 客户端心跳（JavaScript） ​

8. 实战：在线客服系统 + 未读消息 + Redis 集群广播 ​

8.1 系统设计 ​

8.2 实体模型 ​

8.3 客服控制器 ​

8.4 未读消息管理（Redis Hash） ​

8.5 集群版在线状态管理 ​

8.6 WebSocket 配置 ​

8.7 前端集成（JavaScript + STOMP） ​

8.8 Maven 依赖 ​

9. 总结 ​

9.1 技术选型 ​

9.2 关键要点 ​

9.3 参考资料 ​

WebSocket 深度

概述

1. WebSocketHandler 接口源码

1.1 接口定义

1.2 AbstractWebSocketHandler

1.3 自定义 Handler

1.4 WebSocketSession 核心方法

1.5 配置 WebSocket 端点

2. STOMP 协议详解

2.1 什么是 STOMP

2.2 帧格式

2.3 订阅/发布模型

2.4 握手流程

2.5 Spring 配置 STOMP

3. @MessageMapping 注解处理

3.1 基本用法

3.2 支持的参数类型

3.3 路径映射与通配符

3.4 异常处理

4. 用户队列（@SendToUser / UserDestinationResolver）

4.1 @SendToUser 基础

4.2 底层原理：UserDestinationResolver

4.3 手动发送（SimpMessagingTemplate）

5. 广播与点对点通信

5.1 广播实现

5.2 点对点通信

5.3 在线状态管理

6. 集群广播（Redis Pub/Sub + WebSocket 集群）

6.1 问题背景

6.2 架构

6.3 Redis 集群代理实现

6.4 Redis Pub/Sub 配置

7. 心跳保活（Heartbeat）

7.1 原理

7.2 Spring 心跳配置

7.3 客户端心跳（JavaScript）

8. 实战：在线客服系统 + 未读消息 + Redis 集群广播

8.1 系统设计

8.2 实体模型

8.3 客服控制器

8.4 未读消息管理（Redis Hash）

8.5 集群版在线状态管理

8.6 WebSocket 配置

8.7 前端集成（JavaScript + STOMP）

8.8 Maven 依赖

9. 总结

9.1 技术选型

9.2 关键要点

9.3 参考资料