最近在项目中在做一个消息推送的功能,比如客户下单之后通知给给对应的客户发送系统通知,这种消息推送需要使用到全双工的websocket推送消息。

上一篇文章Spring Boot 整合单机websocket介绍了websocket在单机模式下进行消息的发送和接收:

Websocket集群解决方案-LMLPHP

用户A用户Bweb服务器建立连接之后,用户A发送一条消息到服务器,服务器再推送给用户B,在单机系统上所有的用户都和同一个服务器建立连接,所有的session都存储在同一个服务器中。

单个服务器是无法支撑几万人同时连接同一个服务器,需要使用到分布式或者集群将请求连接负载均衡到到不同的服务下。消息的发送方和接收方在同一个服务器,这就和单体服务器类似,能成功接收到消息:

Websocket集群解决方案-LMLPHP

但负载均衡使用轮询的算法,无法保证消息发送方和接收方处于同一个服务器,当发送方和接收方不是在同一个服务器时,接收方是无法接受到消息的:

Websocket集群解决方案-LMLPHP

websocket集群问题解决思路

客户端和服务端每次建立连接时候,会创建有状态的会话session,服务器的保存维持连接的session。客户端每次只能和集群服务器其中的一个服务器连接,后续也是和该服务器进行数据传输。

要解决集群的问题,应该考虑session共享的问题,客户端成功连接服务器之后,其他服务器也知道客户端连接成功。

方案一:session 共享(不可行)

websocket类似的http是如何解决集群问题的?解决方案之一就是共享session,客户端登录服务端之后,将session信息存储在Redis数据库中,连接其他服务器时,从Redis获取session,实际就是将session信息存储在Redis中,实现redis的共享。

session可以被共享的前提是可以被序列化,而websocketsession是无法被序列化的,httpsession记录的是请求的数据,而websocketsession对应的是连接,连接到不同的服务器,session也不同,无法被序列化。

方案二:ip hash(不可行)

http不使用session共享,就可以使用Nginx负载均衡的ip hash算法,客户端每次都是请求同一个服务器,客户端的session都保存在服务器上,而后续请求都是请求该服务器,都能获取到session,就不存在分布式session问题了。

websocket相对http来说,可以由服务端主动推动消息给客户端,如果接收消息的服务端和发送消息消息的服务端不是同一个服务端,发送消息的服务端无法找到接收消息对应的session,即两个session不处于同一个服务端,也就无法推送消息。如下图所示:

Websocket集群解决方案-LMLPHP

方案三:广播模式

将消息的发送方和接收方都处于同一个服务器下才能发送消息,那么可以转换一下思路,可以将消息以消息广播的方式通知给所有的服务器,可以使用消息中间件发布订阅模式,消息脱离了服务器的限制,通过发送到中间件,再发送给订阅的服务器,类似广播一样,只要订阅了消息,都能接收到消息的通知:

Websocket集群解决方案-LMLPHP

发布者发布消息到消息中间件,消息中间件再将发送给所有订阅者:

Websocket集群解决方案-LMLPHP

广播模式的实现

搭建单机 websocket

参考以前写的websocket单机搭建 文章,先搭建单机websocket实现消息的推送。

1. 添加依赖

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

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

2. 创建 ServerEndpointExporter 的 bean 实例

ServerEndpointExporter 的 bean 实例自动注册 @ServerEndpoint 注解声明的 websocket endpoint,使用springboot自带tomcat启动需要该配置,使用独立 tomcat 则不需要该配置。

@Configuration
public class WebSocketConfig {
    //tomcat启动无需该配置
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

3. 创建服务端点 ServerEndpoint 和 客户端端

  • 服务端点
@Component
@ServerEndpoint(value = "/message")
@Slf4j
public class WebSocket {

	private static Map<String, WebSocket> webSocketSet = new ConcurrentHashMap<>();

	private Session session;

	@OnOpen
	public void onOpen(Session session) throws SocketException {
		this.session = session;
		webSocketSet.put(this.session.getId(),this);

		log.info("【websocket】有新的连接,总数:{}",webSocketSet.size());
	}

	@OnClose
	public void onClose(){
		String id = this.session.getId();
		if (id != null){
			webSocketSet.remove(id);
			log.info("【websocket】连接断开:总数:{}",webSocketSet.size());
		}
	}

	@OnMessage
	public void onMessage(String message){
		if (!message.equals("ping")){
			log.info("【wesocket】收到客户端发送的消息,message={}",message);
			sendMessage(message);
		}
	}

	/**
	 * 发送消息
	 * @param message
	 * @return
	 */
	public void sendMessage(String message){
		for (WebSocket webSocket : webSocketSet.values()) {
			webSocket.session.getAsyncRemote().sendText(message);
		}
		log.info("【wesocket】发送消息,message={}", message);

	}

}
  • 客户端点
<div>
    <input type="text" name="message" id="message">
    <button id="sendBtn">发送</button>
</div>
<div style="width:100px;height: 500px;" id="content">
</div>
<script src="https://cdn.bootcdn.net/ajax/libs/jquery/3.6.0/jquery.js"></script>
<script type="text/javascript">
    var ws = new WebSocket("ws://127.0.0.1:8080/message");
    ws.onopen = function(evt) {
        console.log("Connection open ...");
    };

    ws.onmessage = function(evt) {
        console.log( "Received Message: " + evt.data);
        var p = $("<p>"+evt.data+"</p>")
        $("#content").prepend(p);
        $("#message").val("");
    };

    ws.onclose = function(evt) {
        console.log("Connection closed.");
    };

    $("#sendBtn").click(function(){
        var aa = $("#message").val();
        ws.send(aa);
    })

</script>

服务端和客户端中的OnOpenoncloseonmessage都是一一对应的。

  • 服务启动后,客户端ws.onopen调用服务端的@OnOpen注解的方法,储存客户端的session信息,握手建立连接。
  • 客户端调用ws.send发送消息,对应服务端的@OnMessage注解下面的方法接收消息。
  • 服务端调用session.getAsyncRemote().sendText发送消息,对应的客户端ws.onmessage接收消息。

添加 controller

@GetMapping({"","index.html"})
public ModelAndView index() {
	ModelAndView view = new ModelAndView("index");
	return view;
}

效果展示

打开两个客户端,其中的一个客户端发送消息,另一个客户端也能接收到消息。

Websocket集群解决方案-LMLPHP

添加 RabbitMQ 中间件

这里使用比较常用的RabbitMQ作为消息中间件,而RabbitMQ支持发布订阅模式

Websocket集群解决方案-LMLPHP

添加消息订阅

交换机使用扇形交换机,消息分发给每一条绑定该交换机的队列。以服务器所在的IP + 端口作为唯一标识作为队列的命名,启动一个服务,使用队列绑定交换机,实现消息的订阅:

@Configuration
public class RabbitConfig {

    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("PUBLISH_SUBSCRIBE_EXCHANGE");
    }

    @Bean
    public Queue psQueue() throws SocketException {
        // ip + 端口 为队列名 
        String ip = IpUtils.getServerIp() + "_" + IpUtils.getPort();
        return new Queue("ps_" + ip);
    }

    @Bean
    public Binding routingFirstBinding() throws SocketException {
        return BindingBuilder.bind(psQueue()).to(fanoutExchange());
    }
}

修改服务端点 ServerEndpoint

WebSocket添加消息的接收方法,@RabbitListener 接收消息,队列名称使用常量命名,动态队列名称使用 #{name},其中的nameQueuebean 名称:

@RabbitListener(queues= "#{psQueue.name}")
public void pubsubQueueFirst(String message) {
  System.out.println(message);
  sendMessage(message);
}

然后再调用sendMessage方法发送给所在连接的客户端。

修改消息发送

WebSocket类的onMessage方法将消息发送改成RabbitMQ方式发送:

@OnMessage
public void onMessage(String message){
  if (!message.equals("ping")){
    log.info("【wesocket】收到客户端发送的消息,message={}",message);
    //sendMessage(message);
    if (rabbitTemplate == null) {
      rabbitTemplate = (RabbitTemplate) SpringContextUtil.getBean("rabbitTemplate");
    }
    rabbitTemplate.convertAndSend("PUBLISH_SUBSCRIBE_EXCHANGE", null, message);
  }
}

消息通知流程如下所示:

Websocket集群解决方案-LMLPHP

启动两个实例,模拟集群环境

打开idea的Edit Configurations

Websocket集群解决方案-LMLPHP

点击左上角的COPY,然后添加端口server.port=8081

Websocket集群解决方案-LMLPHP

启动两个服务,端口分别是80808081。在启动8081端口的服务,将前端连接端口改成8081:

var ws = new WebSocket("ws://127.0.0.1:8081/message");

效果展示

Websocket集群解决方案-LMLPHP

源码

github源码

参考

11-10 00:04