spring-websocket

Spring WebSocket

快速入门

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

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        // Prefix for messages from server to clients (subscribe)
        registry.enableSimpleBroker("/topic", "/queue");
        // Prefix for messages from clients to server
        registry.setApplicationDestinationPrefixes("/app");
        // Prefix for private messages to a specific user
        registry.setUserDestinationPrefix("/user");
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws")
            .setAllowedOrigins("http://localhost:3000")
            .withSockJS();  // Fallback for browsers without WebSocket
    }
}

---

消息控制器

@Controller
@RequiredArgsConstructor
@Slf4j
public class ChatController {

    private final SimpMessagingTemplate messagingTemplate;

    // Receives message and broadcasts to all subscribers of /topic/chat
    @MessageMapping("/chat.send")
    @SendTo("/topic/chat")
    public ChatMessage sendMessage(ChatMessage message, Principal principal) {
        message.setSender(principal.getName());
        message.setTimestamp(Instant.now());
        return message;
    }

    // Direct reply to the sender
    @MessageMapping("/chat.echo")
    @SendToUser("/queue/reply")
    public ChatMessage echoMessage(ChatMessage message) {
        message.setContent("Echo: " + message.getContent());
        return message;
    }

    // Programmatic send to a specific user
    @MessageMapping("/chat.private")
    public void sendPrivateMessage(PrivateMessage message, Principal principal) {
        message.setSender(principal.getName());
        messagingTemplate.convertAndSendToUser(
            message.getRecipient(),
            "/queue/private",
            message
        );
    }

    // Broadcast to all
    public void broadcastNotification(NotificationMessage notification) {
        messagingTemplate.convertAndSend("/topic/notifications", notification);
    }
}

// DTOs
public record ChatMessage(
    String id, String sender, String content,
    Instant timestamp, MessageType type
) {}

public enum MessageType { CHAT, JOIN, LEAVE, TYPING }

public record PrivateMessage(
    String sender, String recipient, String content, Instant timestamp
) {}

---

事件处理器

@Component
@RequiredArgsConstructor
@Slf4j
public class WebSocketEventListener {

    private final SimpMessagingTemplate messagingTemplate;
    private final OnlineUserService onlineUserService;

    @EventListener
    public void handleSessionConnected(SessionConnectedEvent event) {
        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(event.getMessage());
        String sessionId = accessor.getSessionId();
        Principal principal = accessor.getUser();

        if (principal != null) {
            String username = principal.getName();
            onlineUserService.userConnected(username, sessionId);
            messagingTemplate.convertAndSend("/topic/users.online",
                new UserStatusMessage(username, UserStatus.ONLINE));
        }
    }

    @EventListener
    public void handleSessionDisconnect(SessionDisconnectEvent event) {
        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(event.getMessage());
        String sessionId = accessor.getSessionId();

        onlineUserService.findBySessionId(sessionId).ifPresent(username -> {
            onlineUserService.userDisconnected(sessionId);
            messagingTemplate.convertAndSend("/topic/users.online",
                new UserStatusMessage(username, UserStatus.OFFLINE));
        });
    }
}

完整参考：有关完整的安全配置和验证，请参阅 security.md。

---

安全要点

@Configuration
@EnableWebSocketSecurity
public class WebSocketSecurityConfig {

    @Bean
    public AuthorizationManager<Message<?>> messageAuthorizationManager(
            MessageMatcherDelegatingAuthorizationManager.Builder messages) {
        return messages
            .nullDestMatcher().permitAll()
            .simpSubscribeDestMatchers("/topic/public/**").permitAll()
            .simpSubscribeDestMatchers("/topic/**", "/queue/**").authenticated()
            .simpDestMatchers("/app/**").authenticated()
            .anyMessage().authenticated()
            .build();
    }
}

完整参考：有关 JWT 身份验证、CSRF 防护和消息验证，请参阅 security.md。

---

会话属性和标头

@Controller
public class ChatController {

    @MessageMapping("/chat.join")
    @SendTo("/topic/chat")
    public ChatMessage joinChat(
            @Payload JoinRequest request,
            @Header("simpSessionId") String sessionId,
            SimpMessageHeaderAccessor headerAccessor) {

        // Save attributes in the WebSocket session
        headerAccessor.getSessionAttributes().put("username", request.username());
        headerAccessor.getSessionAttributes().put("roomId", request.roomId());

        return new ChatMessage(null, request.username(),
            request.username() + " joined!", Instant.now(), MessageType.JOIN);
    }
}

---

错误处理

@ControllerAdvice
public class WebSocketExceptionHandler {

    @MessageExceptionHandler
    @SendToUser("/queue/errors")
    public ErrorMessage handleException(Exception e) {
        return new ErrorMessage("ERROR", e.getMessage(), Instant.now());
    }

    @MessageExceptionHandler(AccessDeniedException.class)
    @SendToUser("/queue/errors")
    public ErrorMessage handleAccessDenied(AccessDeniedException e) {
        return new ErrorMessage("ACCESS_DENIED",
            "You don't have permission", Instant.now());
    }
}

public record ErrorMessage(String code, String message, Instant timestamp) {}

---

心跳配置

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableSimpleBroker("/topic", "/queue")
            .setHeartbeatValue(new long[]{10000, 10000})  // Server, Client in ms
            .setTaskScheduler(heartBeatScheduler());
        registry.setApplicationDestinationPrefixes("/app");
    }

    @Bean
    public TaskScheduler heartBeatScheduler() {
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setPoolSize(1);
        scheduler.setThreadNamePrefix("ws-heartbeat-");
        scheduler.initialize();
        return scheduler;
    }

    @Override
    public void configureWebSocketTransport(WebSocketTransportRegistration registry) {
        registry.setMessageSizeLimit(128 * 1024)      // 128KB max message
            .setSendBufferSizeLimit(512 * 1024)       // 512KB send buffer
            .setSendTimeLimit(20 * 1000);             // 20s send timeout
    }
}

完整参考：有关 RabbitMQ 和 Redis 外部代理配置，请参阅 brokers.md。

---

最佳实践

应做	不应做
使用 STOMP + SockJS 兼容不同浏览器	仅使用原始 WebSocket
实现心跳以检测断开连接	依赖 TCP keepalive
使用外部代理（RabbitMQ）进行扩展	在生产环境使用简单代理
处理前验证负载	信任客户端输入
妥善处理断开连接	仅在内存中保存状态

---

不应使用此技能的情况

• REST API - 使用 spring-rest 技能
• 简单 SSE - 使用 SseEmitter 端点
• NestJS WebSocket - 使用 nestjs-websocket 技能
• React 客户端 - 前端使用 react-websocket 技能

---

反模式

反模式	问题	解决方案
连接被拒绝	未配置端点	验证 registerStompEndpoints
403 Forbidden	未配置 CORS	添加 setAllowedOrigins
没有会话	Principal 为 null	配置 WebSocket 身份验证
生产环境使用简单代理	无法水平扩展	使用 RabbitMQ/Redis 适配器
状态保存在内存中	重启后丢失	使用外部会话存储

---

快速故障排查

问题	诊断	修复
连接被拒绝	检查端点配置	验证 registerStompEndpoints
403 Forbidden	检查 CORS	添加 setAllowedOrigins
Principal 为 null	检查身份验证配置	实现 AuthChannelInterceptor
消息过大	检查大小限制	增大 setMessageSizeLimit
心跳超时	检查间隔	正确配置心跳

---

参考文件

文件	内容
brokers.md	RabbitMQ、Redis 外部代理配置
security.md	JWT 身份验证、CSRF、消息验证
advanced.md	底层 WebSocket、JS 客户端、测试

---

外部文档

• Spring WebSocket
• STOMP Protocol