WebSocket详解及Spring Boot+Java高级实战用法示例

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

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        // 注册 STOMP 端点，允许跨域
        registry.addEndpoint("/ws")
                .setAllowedOriginPatterns("*")
                .withSockJS(); // 兼容不支持 WebSocket 的浏览器（降级到 xhr-streaming）
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        // 客户端订阅前缀
        registry.enableSimpleBroker("/topic", "/queue"); // 服务端推送前缀
        registry.setApplicationDestinationPrefixes("/app"); // 客户端发送消息前缀
    }
}

@Controller
public class ChatController {

    @MessageMapping("/chat.sendMessage") // 客户端发送到 /app/chat.sendMessage
    @SendTo("/topic/public")             // 广播给所有订阅 /topic/public 的客户端
    public ChatMessage sendMessage(ChatMessage chatMessage) {
        return chatMessage;
    }

    @MessageMapping("/chat.addUser")
    @SendTo("/topic/public")
    public ChatMessage addUser(@Payload ChatMessage chatMessage,
                               SimpMessageHeaderAccessor headerAccessor) {
        // 将用户名加入会话
        headerAccessor.getSessionAttributes().put("username", chatMessage.getSender());
        chatMessage.setContent("加入聊天室");
        return chatMessage;
    }
}

@Data
@NoArgsConstructor
@AllArgsConstructor
public class ChatMessage {
    private String content;
    private String sender;
    private MessageType type; // enum: CHAT, JOIN, LEAVE

    public enum MessageType {
        CHAT, JOIN, LEAVE
    }
}

<script src="https://cdn.jsdelivr.net/npm/sockjs-client@1/dist/sockjs.min.js"></script>
<script src="https://cdn.jsdelivr.net/npm/stompjs@2.3.3/lib/stomp.min.js"></script>

<script>
const stompClient = Stomp.over(new SockJS('/ws'));

stompClient.connect({}, () => {
    console.log('连接成功');

    // 订阅公共频道
    stompClient.subscribe('/topic/public', (payload) => {
        const msg = JSON.parse(payload.body);
        document.getElementById('chat').innerHTML += 
            `<div><b>${msg.sender}</b>: ${msg.content}</div>`;
    });

    // 发送消息
    function sendMessage() {
        const content = document.getElementById('message').value;
        stompClient.send("/app/chat.sendMessage", {}, JSON.stringify({
            content: content,
            sender: '张三',
            type: 'CHAT'
        }));
    }
});
</script>

// 获取当前登录用户（需集成 Spring Security）
private String getCurrentUser(SimpMessageHeaderAccessor headerAccessor) {
    // 从 token / session 中解析
    return (String) headerAccessor.getSessionAttributes().get("username");
}

@MessageMapping("/chat.private")
public void sendPrivateMessage(@Payload ChatMessage message,
                               SimpMessageHeaderAccessor headerAccessor) {
    String currentUser = getCurrentUser(headerAccessor);
    String recipient = message.getRecipient(); // 假设消息中包含接收者

    // 构造用户专属队列路径
    String destination = "/queue/messages/" + recipient;
    
    // 使用 convertAndSendToUser 发送（自动加 /user 前缀）
    messagingTemplate.convertAndSendToUser(recipient, "/queue/messages", message);
}

// 前端订阅私信
stompClient.subscribe('/user/queue/messages', (payload) => {
    const msg = JSON.parse(payload.body);
    // 显示私聊消息
});

@Component
public class WebSocketEventListener {

    private static final Set<String> onlineUsers = ConcurrentHashMap.newKeySet();

    @EventListener
    public void handleWebSocketConnectListener(SessionConnectedEvent event) {
        // 用户上线（需结合认证）
        System.out.println("用户连接: " + event.getMessage().getHeaders());
    }

    @EventListener
    public void handleWebSocketDisconnectListener(SessionDisconnectEvent event) {
        StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(event.getMessage());
        String username = (String) headerAccessor.getSessionAttributes().get("username");
        if (username != null) {
            onlineUsers.remove(username);
            // 广播在线人数
            messagingTemplate.convertAndSend("/topic/online", onlineUsers.size());
        }
    }
}

public void sendPrivateMessage(ChatMessage message) {
    if (isUserOnline(message.getRecipient())) {
        messagingTemplate.convertAndSendToUser(...);
    } else {
        offlineMessageService.save(message); // 存 DB
    }
}

// 用户上线时
@EventListener
public void onUserOnline(SessionConnectedEvent event) {
    String userId = getCurrentUserId(event);
    List<ChatMessage> offlineMsgs = offlineMessageService.getAndDelete(userId);
    offlineMsgs.forEach(msg -> 
        messagingTemplate.convertAndSendToUser(userId, "/queue/offline", msg)
    );
}

public class HttpHandshakeInterceptor implements HandshakeInterceptor {
    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response,
                                   WebSocketHandler wsHandler, Map<String, Object> attributes) {
        if (request instanceof ServletServerHttpRequest) {
            ServletServerHttpRequest servletRequest = (ServletServerHttpRequest) request;
            String token = servletRequest.getServletRequest().getParameter("token");
            if (isValidToken(token)) {
                attributes.put("userId", parseUserId(token));
                return true;
            }
        }
        return false; // 拒绝连接
    }

    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response,
                               WebSocketHandler wsHandler, Exception exception) {}
}

registry.addEndpoint("/ws")
        .addInterceptors(new HttpHandshakeInterceptor())
        .setAllowedOriginPatterns("*");