SpringBoot+Redis实现分布式WebSocket

什么是分布式WebSocket？

是指在分布式系统架构中实现WebSocket的通信机制，它允许在不同的服务器节点之间共享和同步WebSocket会话状态，从而实现跨多个服务器的实时消息传递。

在分布式环境中实现WebSocket的挑战主要包括以下几点：

会话共享：在分布式系统中，用户的WebSocket连接可能与不同的服务器建立，这就要求系统能够在不同服务器间共享WebSocket会话信息，以便消息能够被正确地传递到所有相关的客户端。负载均衡：使用负载均衡可以提高系统的可用性和伸缩性。但是，当WebSocket请求在服务器之间负载均衡时，需要确保客户端可以与正确的服务器建立连接，并且能够接收到所有的消息。故障转移：在出现服务器故障时，系统需要能够将WebSocket会话无缝迁移到其他健康的服务器上，以保证服务的连续性。一致性：确保所有用户在任何时候看到的都是一致的消息状态，这对于实时通信非常重要。

为了解决这些挑战，可以采取以下几种策略：

使用消息代理：通过引入一个中心化的消息代理（如RabbitMQ、Redis Pub/Sub等），可以让所有的服务器都连接到这个消息代理。当一个服务器需要发送消息时，它将消息发送到消息代理，然后由消息代理负责将消息分发到所有连接的客户端。这样可以确保消息的一致性和可靠性。共享会话存储：使用一个共享的会话存储（如数据库或内存数据网格）来保存WebSocket会话的状态。这样，即使客户端最初连接到的服务器发生故障，其他服务器也可以接管会话并继续处理消息。基于路由的负载均衡：使用智能负载均衡器（如Nginx、HAProxy等），它们可以根据特定的路由规则（如会话ID或用户ID）将WebSocket连接定向到特定的服务器。服务发现：在微服务架构中，可以使用服务发现机制来动态地找到负责特定会话的服务器，并将消息路由到那里。WebSocket代理：使用专门的WebSocket代理服务器，它可以在多个后端服务器之间代理WebSocket连接，并确保消息的传递和会话的同步。应用层协议：设计应用层协议来处理分布式WebSocket的复杂性，例如通过引入心跳机制来检测连接的健康状况，并通过预定的协议来同步会话状态。

总的来说，在实践中，可能需要结合多种策略来构建一个健壮的分布式WebSocket解决方案，以满足不同场景下的需求。此外，还需要考虑安全性、性能和可扩展性等因素，以确保系统的稳定性和可靠性。

温故而知新：单点WebSocket实现

SpringBoot2.0集成WebSocket，实现后台向前端推送信息_springboot集成websocket-CSDN博客https://zhengkai.blog.csdn.net/article/details/80275084

简单版本：在Java中使用Redis实现WebSocket

要在Java中使用Redis实现WebSocket，你需要使用一个支持WebSocket的Java Web框架，如Spring Boot，以及一个支持Redis的Java库，如Jedis。以下是一个简单的示例：

添加依赖项到你的pom.xml文件

org.springframework.boot

spring-boot-starter-websocket

org.springframework.boot

spring-boot-starter-data-redis

创建一个WebSocket配置类

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;

import org.springframework.data.redis.core.RedisTemplate;

import org.springframework.data.redis.listener.ChannelTopic;

import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;

import org.springframework.web.socket.config.annotation.StompEndpointRegistry;

import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;

//by zhengkai.blog.csdn.net

@Configuration

@EnableWebSocketMessageBroker

public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

@Override

public void registerStompEndpoints(StompEndpointRegistry registry) {

registry.addEndpoint("/websocket").withSockJS();

}

@Override

public void configureMessageBroker(org.springframework.messaging.simp.config.MessageBrokerRegistry registry) {

registry.enableSimpleBroker("/topic");

registry.setApplicationDestinationPrefixes("/app");

}

@Bean

public JedisConnectionFactory jedisConnectionFactory() {

return new JedisConnectionFactory();

}

@Bean

public RedisTemplate redisTemplate() {

RedisTemplate template = new RedisTemplate<>();

template.setConnectionFactory(jedisConnectionFactory());

return template;

}

@Bean

public MessageListenerAdapter messageListenerAdapter() {

return new MessageListenerAdapter(new RedisMessageListener());

}

@Bean

public RedisMessageListenerContainer redisMessageListenerContainer() {

RedisMessageListenerContainer container = new RedisMessageListenerContainer();

container.setConnectionFactory(jedisConnectionFactory());

container.addMessageListener(messageListenerAdapter(), topic());

return container;

}

@Bean

public ChannelTopic topic() {

return new ChannelTopic("websocket-topic");

}

}

创建一个WebSocket消息监听器

import org.springframework.data.redis.connection.Message;

import org.springframework.data.redis.connection.MessageListener;

import org.springframework.stereotype.Component;

@Component

public class RedisMessageListener implements MessageListener {

@Override

public void onMessage(Message message, byte[] pattern) {

System.out.println("Received message: " + message);

}

}

发送消息到WebSocket客户端

在你的控制器中，你可以使用SimpMessagingTemplate来发送消息到WebSocket客户端：

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.messaging.simp.SimpMessagingTemplate;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RestController;

@RestController

public class WebSocketController {

@Autowired

private SimpMessagingTemplate messagingTemplate;

@GetMapping("/send")

public String sendMessage() {

messagingTemplate.convertAndSend("/topic/websocket-topic", "Hello from Redis!");

return "Message sent!";

}

}

正式版本：用SpringBoot+Redis实现分布式WebSocket

将消息（<用户id，消息内容>）统一推送到一个消息队列（Redis、Kafka等）的的topic，然后每个应用节点都订阅这个topic，在接收到WebSocket消息后取出这个消息的“消息接收者的用户ID/用户名”，然后再比对自身是否存在相应用户的连接，如果存在则推送消息，否则丢弃接收到的这个消息（这个消息接收者所在的应用节点会处理）在用户建立WebSocket连接后，使用Redis缓存记录用户的WebSocket建立在哪个应用节点上，然后同样使用消息队列将消息推送到接收者所在的应用节点上面（实现上比方案一要复杂，但是网络流量会更低）

 1. 定义一个WebSocket Channel枚举类

public enum WebSocketChannelEnum {

//测试使用的简易点对点聊天

CHAT("CHAT", "测试使用的简易点对点聊天", "/topic/reply");

WebSocketChannelEnum(String code, String description, String subscribeUrl) {

this.code = code;

this.description = description;

this.subscribeUrl = subscribeUrl;

}

/**

* 唯一CODE

*/

private String code;

/**

* 描述

*/

private String description;

/**

* WebSocket客户端订阅的URL

*/

private String subscribeUrl;

public String getCode() {

return code;

}

public String getDescription() {

return description;

}

public String getSubscribeUrl() {

return subscribeUrl;

}

/**

* 通过CODE查找枚举类

*/

public static WebSocketChannelEnum fromCode(String code){

if(StringUtils.isNoneBlank(code)){

for(WebSocketChannelEnum channelEnum : values()){

if(channelEnum.code.equals(code)){

return channelEnum;

}

}

}

return null;

}

}

2. 配置基于Redis的消息队列

需要注意的是，在大中型正式项目中并不推荐使用Redis实现的消息队列，因为经过测试它并不是特别可靠，所以应该考虑使用Kafka、rabbitMQ等专业的消息队列中间件

@Configuration

@ConditionalOnClass({JedisCluster.class})

public class RedisConfig {

@Value("${spring.redis.timeout}")

private String timeOut;

@Value("${spring.redis.cluster.nodes}")

private String nodes;

@Value("${spring.redis.cluster.max-redirects}")

private int maxRedirects;

@Value("${spring.redis.jedis.pool.max-active}")

private int maxActive;

@Value("${spring.redis.jedis.pool.max-wait}")

private int maxWait;

@Value("${spring.redis.jedis.pool.max-idle}")

private int maxIdle;

@Value("${spring.redis.jedis.pool.min-idle}")

private int minIdle;

@Value("${spring.redis.message.topic-name}")

private String topicName;

@Bean

public JedisPoolConfig jedisPoolConfig(){

JedisPoolConfig config = new JedisPoolConfig();

config.setMaxTotal(maxActive);

config.setMaxIdle(maxIdle);

config.setMinIdle(minIdle);

config.setMaxWaitMillis(maxWait);

return config;

}

@Bean

public RedisClusterConfiguration redisClusterConfiguration(){

RedisClusterConfiguration configuration = new RedisClusterConfiguration(Arrays.asList(nodes));

configuration.setMaxRedirects(maxRedirects);

return configuration;

}

/**

* JedisConnectionFactory

*/

@Bean

public JedisConnectionFactory jedisConnectionFactory(RedisClusterConfiguration configuration,JedisPoolConfig jedisPoolConfig){

return new JedisConnectionFactory(configuration,jedisPoolConfig);

}

/**

* 使用Jackson序列化对象

*/

@Bean

public Jackson2JsonRedisSerializer