第 72 章：WebSocket 与实时通信

学习目标

理解 WebSocket 协议与 HTTP 的区别
掌握 Spring Boot WebSocket 集成
实现群消息推送、点对点聊天、实时通知

一、为什么需要 WebSocket？

方案	原理	实时性	服务器压力
轮询	客户端定时请求	差（看轮询间隔）	高
长轮询	请求挂起等消息	中	中
SSE	服务端单向推送	中	低
WebSocket	全双工双向通信	最强	中

二、Spring Boot 集成 WebSocket

xml

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

配置

java

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(new ChatWebSocketHandler(), "/ws/chat")
                .setAllowedOrigins("*");
    }
}

处理器

java

@Slf4j
@Component
public class ChatWebSocketHandler extends TextWebSocketHandler {

    // ① 维护所有在线连接
    private static final Map<Long, WebSocketSession> SESSIONS = new ConcurrentHashMap<>();

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        // ② 从 session 属性中取用户 ID（在握手时设置）
        Long userId = (Long) session.getAttributes().get("userId");
        if (userId != null) {
            SESSIONS.put(userId, session);
            log.info("用户 {} 上线，当前在线 {}", userId, SESSIONS.size());
        }
    }

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message)
            throws Exception {
        // ③ 收到消息：解析、转发
        ChatMessage msg = JSON.parseObject(message.getPayload(), ChatMessage.class);

        if (msg.getToUserId() != null) {
            // ④ 点对点消息
            sendToUser(msg.getToUserId(), msg);
        } else if (msg.getRoomId() != null) {
            // ⑤ 群消息
            sendToRoom(msg.getRoomId(), msg);
        }
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        Long userId = (Long) session.getAttributes().get("userId");
        SESSIONS.remove(userId);
        log.info("用户 {} 下线", userId);
    }

    // ⑥ 点对点发送
    public void sendToUser(Long userId, Object message) throws IOException {
        WebSocketSession session = SESSIONS.get(userId);
        if (session != null && session.isOpen()) {
            session.sendMessage(new TextMessage(JSON.toJSONString(message)));
        }
    }

    // ⑦ 群发（实际项目维护 roomId → userIds 映射）
    public void sendToRoom(Long roomId, Object message) throws IOException {
        List<Long> userIds = roomService.getMembers(roomId);
        for (Long userId : userIds) {
            sendToUser(userId, message);
        }
    }
}

握手拦截器（鉴权）

java

@Component
@Slf4j
public class AuthHandshakeInterceptor implements HandshakeInterceptor {

    private final JwtUtil jwtUtil;

    @Override
    public boolean beforeHandshake(ServerHttpRequest request,
                                   ServerHttpResponse response,
                                   WebSocketHandler wsHandler,
                                   Map<String, Object> attributes) {
        // ① 从 URL 参数或 Header 拿 token
        String token = extractToken(request);
        try {
            Claims claims = jwtUtil.parse(token);
            Long userId = Long.valueOf(claims.getSubject());
            attributes.put("userId", userId);          // 存到 session 属性
            return true;                              // 允许连接
        } catch (Exception e) {
            response.setStatusCode(HttpStatus.UNAUTHORIZED);
            return false;
        }
    }

    @Override
    public void afterHandshake(...) { }

    private String extractToken(ServerHttpRequest request) {
        // 从 Query 参数拿：ws://localhost:8080/ws/chat?token=xxx
        URI uri = request.getURI();
        String query = uri.getQuery();
        for (String pair : query.split("&")) {
            String[] kv = pair.split("=");
            if (kv.length == 2 && "token".equals(kv[0])) {
                return kv[1];
            }
        }
        return null;
    }
}

java

@Configuration
@EnableWebSocket
@RequiredArgsConstructor
public class WebSocketConfig implements WebSocketConfigurer {

    private final AuthHandshakeInterceptor authInterceptor;
    private final ChatWebSocketHandler chatHandler;

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(chatHandler, "/ws/chat")
                .addInterceptors(authInterceptor)
                .setAllowedOrigins("*");
    }
}

客户端（JavaScript）

javascript

// 1. 建立连接
const socket = new WebSocket('ws://localhost:8080/ws/chat?token=' + token)

socket.onopen = () => {
    console.log('连接已建立')
}

socket.onmessage = (event) => {
    const msg = JSON.parse(event.data)
    console.log('收到消息:', msg)
    showMessage(msg)
}

socket.onclose = () => {
    console.log('连接已断开，3秒后重连')
    setTimeout(() => connect(), 3000)              // ① 自动重连
}

socket.onerror = (err) => {
    console.error('连接错误:', err)
}

// 2. 发送消息
function send(toUserId, content) {
    socket.send(JSON.stringify({
        type: 'PRIVATE',
        toUserId: toUserId,
        content: content
    }))
}

三、STOMP 协议（更强大）

STOMP（Simple Text Oriented Messaging Protocol）是 WebSocket 之上的子协议，支持主题订阅、消息确认等高级特性。

xml

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

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")
                .setAllowedOrigins("*")
                .withSockJS();                                  // ④ SockJS fallback
    }
}

Controller（用 @MessageMapping）

java

@Controller
public class ChatController {

    private final SimpMessagingTemplate messagingTemplate;

    // 客户端发送到 /app/chat → 路由到这个方法
    @MessageMapping("/chat")
    public void chat(ChatMessage message) {
        // ① 持久化
        chatService.save(message);

        // ② 推送给目标用户
        messagingTemplate.convertAndSendToUser(
                String.valueOf(message.getToUserId()),
                "/queue/messages",
                message);
    }

    @MessageMapping("/join")
    public void join(JoinMessage message) {
        // ③ 加入群组后通知其他人
        messagingTemplate.convertAndSend(
                "/topic/room/" + message.getRoomId(),
                new SystemMessage("系统", message.getUsername() + " 加入了群聊"));
    }
}

客户端订阅

javascript

// 用 SockJS + StompJS
const socket = new SockJS('/ws/stomp')
const stompClient = Stomp.over(socket)

stompClient.connect({}, () => {
    // ① 订阅个人消息队列
    stompClient.subscribe('/user/queue/messages', (msg) => {
        console.log('个人消息:', JSON.parse(msg.body))
    })

    // ② 订阅群消息
    stompClient.subscribe('/topic/room/1001', (msg) => {
        console.log('群消息:', JSON.parse(msg.body))
    })

    // ③ 发送消息
    stompClient.send('/app/chat', {}, JSON.stringify({
        toUserId: 1002,
        content: '你好'
    }))
})

四、分布式场景：WebSocket + Redis

问题：用户 A 连到 server1，用户 B 连到 server2，A 给 B 发消息怎么转发？

java

@Component
@RequiredArgsConstructor
@Slf4j
public class DistributedWebSocketHandler extends TextWebSocketHandler
        implements ApplicationRunner {

    private final RedisTemplate<String, Object> redis;
    private static final Map<Long, WebSocketSession> LOCAL_SESSIONS = new ConcurrentHashMap<>();

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        Long userId = (Long) session.getAttributes().get("userId");
        LOCAL_SESSIONS.put(userId, session);
        log.info("用户 {} 连接到本机", userId);
    }

    // ① 收到 Redis 消息时，处理本机的连接
    public void onRedisMessage(String channel, String message) {
        ChatMessage msg = JSON.parseObject(message, ChatMessage.class);
        WebSocketSession session = LOCAL_SESSIONS.get(msg.getToUserId());
        if (session != null && session.isOpen()) {
            try {
                session.sendMessage(new TextMessage(JSON.toJSONString(msg)));
            } catch (IOException e) {
                log.error("发送失败", e);
            }
        }
    }

    // ② 主动发消息（可能跨机器）
    public void sendMessage(ChatMessage msg) {
        // 先查本机有没有
        WebSocketSession session = LOCAL_SESSIONS.get(msg.getToUserId());
        if (session != null && session.isOpen()) {
            try {
                session.sendMessage(new TextMessage(JSON.toJSONString(msg)));
                return;
            } catch (IOException e) { /* ignore */ }
        }

        // ③ 本机没有，发到 Redis 广播
        redis.convertAndSend("chat:message", JSON.toJSONString(msg));
    }

    @Override
    public void run(ApplicationArguments args) {
        // ④ 订阅 Redis 频道
        redis.listen(RedisListener.of("chat:message", this::onRedisMessage));
    }
}

五、消息推送（服务端主动）

java

@Service
@RequiredArgsConstructor
public class NotificationService {

    private final SimpMessagingTemplate messagingTemplate;

    // ① 系统通知（推送给指定用户）
    public void notifyUser(Long userId, String content) {
        Notification n = new Notification("系统", content, LocalDateTime.now());
        messagingTemplate.convertAndSendToUser(
                String.valueOf(userId),
                "/queue/notifications",
                n);
    }

    // ② 广播通知（推送给所有在线用户）
    public void broadcast(String content) {
        Notification n = new Notification("公告", content, LocalDateTime.now());
        messagingTemplate.convertAndSend(
                "/topic/notifications",
                n);
    }

    // ③ 监听业务事件（订单支付后通知商家）
    @EventListener
    public void onOrderPaid(OrderPaidEvent event) {
        notifyUser(event.getMerchantId(),
                "您的订单 " + event.getOrderId() + " 已支付");
    }
}

六、连接管理

心跳检测

java

// 客户端：每 30 秒发心跳
setInterval(() => {
    if (socket.readyState === WebSocket.OPEN) {
        socket.send('ping')
    }
}, 30000)

// 服务端：超时无心跳则断开
public class HeartbeatHandler extends TextWebSocketHandler {

    private static final long TIMEOUT = 60_000;

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage msg) {
        if ("ping".equals(msg.getPayload())) {
            session.sendMessage(new TextMessage("pong"));
            session.setLastHeartbeat(System.currentTimeMillis());
        }
    }

    @Scheduled(fixedDelay = 10000)
    public void checkTimeout() {
        long now = System.currentTimeMillis();
        SESSIONS.forEach((uid, session) -> {
            Long lastHb = (Long) session.getAttributes().get("lastHeartbeat");
            if (lastHb != null && now - lastHb > TIMEOUT) {
                try {
                    session.close(CloseStatus.SESSION_NOT_RELIABLE);
                } catch (IOException e) { /* ignore */ }
            }
        });
    }
}

在线状态

java

@Service
@RequiredArgsConstructor
public class OnlineStatusService {

    private final RedisTemplate<String, Object> redis;

    public void online(Long userId) {
        redis.opsForValue().set("online:" + userId, "1", 30, TimeUnit.MINUTES);
    }

    public void offline(Long userId) {
        redis.delete("online:" + userId);
    }

    // ① 在线状态心跳续期
    @Scheduled(fixedDelay = 60000)
    public void refreshOnline() {
        for (Long userId : LOCAL_SESSIONS.keySet()) {
            online(userId);
        }
    }

    public boolean isOnline(Long userId) {
        return Boolean.TRUE.equals(redis.hasKey("online:" + userId));
    }
}

七、应用场景

场景	方案
即时聊天	WebSocket + STOMP
实时通知	WebSocket + Redis Pub/Sub
协作编辑	WebSocket + OT/CRDT 算法
实时数据大屏	SSE（服务端单向推送足够）
直播弹幕	WebSocket + 消息队列
在线客服	WebSocket + 客服路由

八、本章小结

要点	关键
WebSocket	全双工双向通信，替代轮询
Spring 集成	`WebSocketHandler` + `WebSocketConfigurer`
STOMP	WebSocket 之上的子协议，支持主题订阅
鉴权	`HandshakeInterceptor` 在握手时校验 JWT
分布式	Redis Pub/Sub 跨服务器转发
心跳	客户端 ping/pong，服务端超时断开
在线状态	Redis 维护 + 心跳续期
注意	浏览器连接数限制（6 个/域名）、需做好重连

动手练习

练习 1：基础题

实现一个简单的实时聊天室：用户登录后连接 WebSocket，可以向特定用户发送消息。

练习 2：进阶题

集成 STOMP，实现群聊：用户加入房间后能收到所有成员的消息，离线消息入库后下次上线推送。

练习 3：思考题

你的系统需要在订单支付后实时通知商家。如何用 Spring Event + WebSocket 实现解耦？

下一章：第 73 章：监控与可观测性 →

第 72 章：WebSocket 与实时通信 ​

一、为什么需要 WebSocket？ ​

二、Spring Boot 集成 WebSocket ​

配置 ​

处理器 ​

握手拦截器（鉴权） ​

客户端（JavaScript） ​

三、STOMP 协议（更强大） ​

Controller（用 @MessageMapping） ​

客户端订阅 ​

四、分布式场景：WebSocket + Redis ​

五、消息推送（服务端主动） ​

六、连接管理 ​

心跳检测 ​

在线状态 ​

七、应用场景 ​

八、本章小结 ​

动手练习 ​

练习 1：基础题 ​

练习 2：进阶题 ​

练习 3：思考题 ​

第 72 章：WebSocket 与实时通信

一、为什么需要 WebSocket？

二、Spring Boot 集成 WebSocket

配置

处理器

握手拦截器（鉴权）

客户端（JavaScript）

三、STOMP 协议（更强大）

Controller（用 @MessageMapping）

客户端订阅

四、分布式场景：WebSocket + Redis

五、消息推送（服务端主动）

六、连接管理

心跳检测

在线状态

七、应用场景

八、本章小结

动手练习

练习 1：基础题

练习 2：进阶题

练习 3：思考题