WebSocket分布式实现方案

版本使用

Springboot使用3.3.1版本

jdk21

实现逻辑

使用Redis记录用户连接地址，RabbitMQ为每个微服务所在的服务器创建对应的交换机或特定的路由规则，每个微服务监听自己的交换机。当连接不同服务器的两个用户进行通信时，通过MQ将消息进行转发到对应的队列中，由监听该队列的微服务获取消息进行转发到连接该服务器的用户。例如：用户A连接到120.78.2.56服务器，用户B连接到120.78.2.57服务器，当用户A要给用户B发送消息时，用户A请求56服务器的接口，将消息发送到56服务器，56服务器上的应用判断用户B的连接是否在自己这里，如果在自己这里则直接进行消息转发即可，如果不在则查询Redis中用户B连接的服务器地址，然后将消息发送到MQ对应的队列或交换机上，由57服务器监听消息，然后发送给连接在57服务器上的用户B。

代码实现

pom依赖

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.creatar</groupId><artifactId>creatar</artifactId><version>1.0</version><packaging>jar</packaging><properties><maven.compiler.source>21</maven.compiler.source><maven.compiler.target>21</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>3.3.1</version></parent><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-undertow</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><!-- https://mvnrepository.com/artifact/io.netty/netty-all --><dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.112.Final</version></dependency><!-- https://mvnrepository.com/artifact/org.postgresql/postgresql --><dependency><groupId>org.postgresql</groupId><artifactId>postgresql</artifactId><version>42.7.4</version></dependency><!-- https://mvnrepository.com/artifact/com.alibaba/druid-spring-boot-3-starter --><dependency><groupId>com.alibaba</groupId><artifactId>druid-spring-boot-3-starter</artifactId><version>1.2.23</version></dependency><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-spring-boot3-starter</artifactId><version>3.5.7</version></dependency><!-- https://mvnrepository.com/artifact/com.github.yulichang/mybatis-plus-join-boot-starter --><dependency><groupId>com.github.yulichang</groupId><artifactId>mybatis-plus-join-boot-starter</artifactId><version>1.4.13</version></dependency><!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-data-redis --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><dependency><groupId>org.redisson</groupId><artifactId>redisson-spring-boot-starter</artifactId><version>3.26.0</version></dependency></dependencies></project>

创建Netty服务器

package com.creatar.service.common.im.core;import com.creatar.properties.SocketProperties;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
/*** Netty服务器** @author: 张定辉* @date: 2025-05-03* @description: Netty服务器*/
@Component
@RequiredArgsConstructor
public class NettyServer implements ApplicationRunner {private static final NioEventLoopGroup bossGroup = new NioEventLoopGroup();private static final NioEventLoopGroup workerGroup = new NioEventLoopGroup();private final SocketProperties socketProperties;private final SocketMessageHandler socketMessageHandler;private Channel serverChannel;private ServerBootstrap bootstrap() {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {socketChannel.pipeline().addLast(new HttpServerCodec());socketChannel.pipeline().addLast(new ChunkedWriteHandler());socketChannel.pipeline().addLast(new HttpObjectAggregator(65536));socketChannel.pipeline().addLast(new WebSocketServerProtocolHandler(socketProperties.getHost()));socketChannel.pipeline().addLast(new LoggingHandler());socketChannel.pipeline().addLast(socketMessageHandler);}}).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true);return serverBootstrap;}@Overridepublic void run(ApplicationArguments args) throws Exception {serverChannel = bootstrap().bind(socketProperties.getPort()).sync().channel().closeFuture().sync().channel();}@PreDestroypublic void stop() {serverChannel.close();serverChannel.parent().close();}
}

用户连接/在线情况业务类

package com.creatar.service.common.im.core;import com.creatar.properties.SocketProperties;
import com.creatar.util.HttpUtil;
import lombok.RequiredArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;/*** 用户在线情况业务类** @author: 张定辉* @date: 2025-05-03* @description: 用户在线情况业务类*/
@Service
@RequiredArgsConstructor
public class OnlineService {private final SocketProperties socketProperties;private final RedisTemplate<String, String> redisTemplate;/*** 设置用户上线** @param userId 用户ID*/public void setUserOnline(String userId) {if (StringUtils.isBlank(userId)) {return;}redisTemplate.opsForValue().setBit(socketProperties.getOnlineKey(), Long.parseLong(userId), true);}/*** 设置用户下线*/public void setUserOffline(String userId) {if (StringUtils.isBlank(userId)) {return;}redisTemplate.opsForValue().setBit(socketProperties.getOnlineKey(), Long.parseLong(userId), false);}/*** 统计在线人数** @return 返回在线人数*/public Long statOnline() {return redisTemplate.execute((RedisCallback<Long>) con -> con.stringCommands().bitCount(socketProperties.getOnlineKey().getBytes()));}/*** 设置用户连接节点*/public void setUserConnectionNode(String userId) {redisTemplate.opsForValue().set(socketProperties.getUserConnectionNode() + userId, HttpUtil.getHostname());redisTemplate.expire(socketProperties.getUserConnectionNode() + userId, 1, TimeUnit