Java上的WebSocket：高级实时应用

Java上的WebSocket：实时应用的高级技术

WebSocket是一种全双工通信协议，允许在客户端和服务器之间建立持久连接。与HTTP每次请求创建新连接不同，WebSocket支持实时连续的数据交换。在Java中，该协议通过JSR 356（Java API for WebSocket）和Spring WebSocket等强大框架实现。本文将探讨Java实时应用开发的高级方面：扩展、集群、与Spring的集成以及复杂场景的处理。

Java WebSocket应用的架构与扩展

在开发Java实时系统时，重要的是要理解WebSocket连接是长期存在且保持状态（stateful）的连接。这会在扩展时带来问题：如果应用运行在多台服务器上，如何确保发送到一台服务器的消息能够到达连接到另一台服务器的客户端？

解决方案：使用消息代理。 流行的解决方案包括Redis Pub/Sub或RabbitMQ。它们充当中央“总线”，所有服务器通过它交换数据。当一台服务器从客户端收到消息时，它会将消息发布到代理，而所有其他服务器订阅该频道并将消息发送给各自连接的客户端。

在Spring上集成Redis Pub/Sub的示例：

@Component\npublic class RedisMessagePublisher implements MessagePublisher {\n    @Autowired\n    private StringRedisTemplate stringRedisTemplate;\n\n    public void publish(String channel, String message) {\n        stringRedisTemplate.convertAndSend(channel, message);\n    }\n}\n\n@Service\npublic class WebSocketRedisSubscriber implements MessageListener {\n    @Autowired\n    private SimpMessagingTemplate messagingTemplate;\n\n    @Override\n    public void onMessage(Message message, byte[] pattern) {\n        String payload = new String(message.getBody());\n        // 发送给所有WebSocket订阅者\n        messagingTemplate.convertAndSend("/topic/public", payload);\n    }\n}

此外，为了扩展，需要正确管理会话。不要将会话存储在单个服务器的内存中（HashMap），而应使用分布式存储，例如Hazelcast或Redis。这将允许集群中的任何服务器访问任何会话。

Spring WebSocket与STOMP的高级应用

Spring WebSocket提供了对JSR 356的高级抽象。在WebSocket之上使用STOMP（简单文本定向消息协议）简化了消息路由和订阅管理。

高级配置：

@Configuration\n@EnableWebSocketMessageBroker\npublic class WebSocketConfig implements WebSocketMessageBrokerConfigurer {\n\n    @Override\n    public void configureMessageBroker(MessageBrokerRegistry config) {\n        // 用于简单场景的内置代理\n        config.enableSimpleBroker("/topic", "/queue");\n        // 从客户端到服务器的消息前缀\n        config.setApplicationDestinationPrefixes("/app");\n        // 使用外部代理（例如RabbitMQ）\n        config.enableStompBrokerRelay("/topic", "/queue")\n              .setRelayHost("localhost")\n              .setRelayPort(61613)\n              .setClientLogin("guest")\n              .setClientPasscode("guest");\n    }\n\n    @Override\n    public void registerStompEndpoints(StompEndpointRegistry registry) {\n        registry.addEndpoint("/ws")\n                .setAllowedOrigins("*")\n                .withSockJS(); // 针对旧浏览器的回退方案\n    }\n}

使用注解处理复杂场景：

\n
• @MessageMapping — 处理来自客户端的消息。
\n