websocket集群的问题及解决方案

websocket集群的问题及解决方案

现在的互联网项目大多采用分布式+微服务+服务集群的方式，那么当项目中的websocket采用集群时就会遇到这么一个问题：

给用户页面推送消息的websocket服务未必是与该用户建立websocket连接的服务。

如果在单机情况下，当websocket需要给用户推送消息时，由于用户已经与websocket服务建立连接，消息推送能够成功。

但如果在集群情况下，用户甲向websocket发起连接请求，有多台服务时，只能与一台服务建立连接（以服务A为例），而这些websocket服务都是有可能会给用户甲推送消息，这时候的服务B和C并没有建立连接，所以会有一部分消息推送失败。

我们通过消息组件来处理这个问题，在原来的系统里，由业务模块的服务处理业务逻辑，发生需要推送给用户的数据的时候，发消息给websocket服务，由websocket服务推送消息给用户。

那么，问题的关键就是确保和用户建立连接的websocket服务就是接收到消息的服务，但是由于建立连接的websockt是由注册中心进行负载均衡分配的，接收消息的服务也是无法确定的，这一点并没有什么太好的办法保证。

因此，我们可以换个思路，只要确保对于业务模块发送的消息，所有的websocket服务都能收到消息，只要做到了这一点，与用户建立连接websocket自然也能接收到消息。而且，这种方式相对单台服务收到消息还有一个在处理多点登陆场景下的优势。对于允许多点登录的系统，同一用户可以在多处进行登录，以为着同一用户与多个服务拥有多个websocket连接，这就要求我们保证多台用户消费同一台业务模块的消息。

现在的问题就变成了每台websocket服务都能收到业务模块发的消息，以rabbitmq为例，只要做到每台服务的queue不同，并都有绑定到同一个TopicExchange。其它的消息组件也是类似，比如kafka需要做到每台服务的consumer group不同。

当然，虽然websocket服务时集群模式，但应该只是同一套代码部署多次，不应该使用不同的代码或者需要额外的人工处理，否则维护难度大并且易出错。所以我们选择使用服务器的ip和端口作为queueName的后缀来保证各个服务的queueName的不同。

在websocket服务模块声明TopicExchange和增加后缀的方法。

@Bean(name = "orderWebSocketExchange")
public TopicExchange orderWebSocketExchange() {
	return new TopicExchange(OrderMqMessage.ORDER_WEBSOCKET_EXCHANGE);
}

@Value("${server.port}")
private String port;

protected String withServiceId(String keyName) {
	String queueName = new String();
	try {
		queueName = keyName + "_" + InetAddress.getLocalHost().getHostAddress() + ":" + port;
	} catch (UnknownHostException e) {
		queueName = keyName + "_" + UUID.randomUUID();
	} 
	return queueName;
}

然后对于每一个webcoket相关的rabbitmq消息，声明队列并绑定到TopicExchange。

@Bean(name = "orderShowQueue")
public Queue orderShowQueue() {
	return new Queue(withServiceId(ORDER_SHOW_QUEUE));
}

@Bean(name = "orderShowQueueBind")
public Binding orderShowQueueBind(@Qualifier("orderShowQueue") Queue queue, TopicExchange topicExchange) {
	returnBindingBuilder.bind(queue).to(topicExchange).with(OrderMqMessage.ORDER_SHOW_PATTERN);
}

另外，我们原先可以通过@RabbitListener注解来实现rabbitmq消息的消费，但是RabbitListener的需要配置queueName作为其queue属性来实现匹配，而现在我们queue加上了服务ip和端口作为后缀，queueName并非是一个常量，因此我们需要换一种方式来实现rabbitmq的监听与消费。

@Bean("orderShowMessageListener")
//使用orderMqConsumer的orderShow方法作为listener。
MessageListenerAdapter orderShowMessageListener(OrderMqConsumer orderMqConsumer) {
    return new MessageListenerAdapter(orderMqConsumer, "orderShow");
}

//在container内将queue和listener绑定
@Bean("orderShowMessageListenerContainer")
SimpleMessageListenerContainer orderShowMessageListenerContainer(ConnectionFactory connectionFactory, 
		@Qualifier("orderShowMessageListener") MessageListenerAdapter listenerAdapter, 
		@Qualifier("orderShowQueue") Queue queue) {
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);
    container.setQueueNames(queue.getName());
    container.setMessageListener(listenerAdapter);
    return container;
}

至此，websocket集群问题得到了解决。

优点：
 1.解决了websockcet集群的消息推送问题。
2.代码的解耦，处理集群问题的代码与消费rabbitmq并推送websocket消息的代码进行了解耦，不需要修改原有的代码逻辑，只需要改变configuration文件的一些配置类。
3.能够做到随意增减服务节点数量，不需要额外修改代码，不需要任何额外的配置。

缺点：
对于每一条与websocket推送相关的rabbitmq消息都需要增加声明listener和在container内将queue和listener绑定的配置。
优化点：
rabbitListener注解根据queue的name进行匹配和消费，我们现在的queue的name是变量，无法使用rabbitListener注解，但我们queue的beanName是能确定的，因此我们可以模仿rabbitListener写一个根据queue的beanName进行匹配和消费的注解。
————————————————
版权声明：本文为CSDN博主「qq_39470742」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/qq_39470742/article/details/83411912

Comments are closed.