單機(jī)節(jié)點(diǎn)下，WebSocket連接成功后，可以直接發(fā)送消息。而多節(jié)點(diǎn)下，連接時(shí)通過(guò)nginx會(huì)代理到不同節(jié)點(diǎn)。

假設(shè)一開(kāi)始用戶(hù)連接了node1的socket服務(wù)。觸發(fā)消息發(fā)送的條件的時(shí)候也通過(guò)nginx進(jìn)行代理，假如代理轉(zhuǎn)到了node2節(jié)點(diǎn)上，那么node2節(jié)點(diǎn)的socket服務(wù)就發(fā)送不了消息，因?yàn)橐婚_(kāi)始用戶(hù)注冊(cè)的是node1節(jié)點(diǎn)。這就導(dǎo)致了消息發(fā)送失敗。

為了解決這一方案，消息發(fā)送時(shí)，就需要一個(gè)中間件來(lái)記錄，這樣，三個(gè)節(jié)點(diǎn)都可以獲取消息，然后在根據(jù)條件進(jìn)行消息推送。

1、依賴(lài)

<!-- redis --> <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><!-- websocket --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-websocket</artifactId></dependency>

2、創(chuàng)建業(yè)務(wù)處理類(lèi) Demo.class，該類(lèi)可以實(shí)現(xiàn)MessageListener接口后重寫(xiě)onMessage方法，也可以不實(shí)現(xiàn)，自己寫(xiě)方法。

import com.alibaba.fastjson.JSON;import com.dy.service.impl.OrdersServiceImpl;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.data.redis.connection.Message;import org.springframework.data.redis.connection.MessageListener;import org.springframework.stereotype.Component; import java.util.HashMap; /** * @program: * @description: redis消息訂閱-業(yè)務(wù)處理 * @author: zhang yi * @create: 2021-01-25 16:46 */@Componentpublic class Demo implements MessageListener { Logger logger = LoggerFactory.getLogger(this.getClass()); @Override public void onMessage(Message message, byte[] pattern) { logger.info('消息訂閱成功---------'); logger.info('內(nèi)容：'+message.getBody()); logger.info('交換機(jī)：'+message.getChannel()); }}

3、創(chuàng)建PubSubConfig配置類(lèi)

import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.cache.annotation.EnableCaching;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.data.redis.connection.RedisConnectionFactory;import org.springframework.data.redis.core.StringRedisTemplate;import org.springframework.data.redis.listener.PatternTopic;import org.springframework.data.redis.listener.RedisMessageListenerContainer;import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; /** * @program: * @description: redis發(fā)布訂閱配置 * @author: zhang yi * @create: 2021-01-25 16:49 */@Configuration@EnableCachingpublic class PubSubConfig { Logger logger = LoggerFactory.getLogger(this.getClass()); //如果是多個(gè)交換機(jī)，則參數(shù)為(RedisConnectionFactory connectionFactory, // MessageListenerAdapter listenerAdapter, // MessageListenerAdapter listenerAdapter2) @Bean RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); // 可以添加多個(gè) messageListener，配置不同的交換機(jī) container.addMessageListener(listenerAdapter, new PatternTopic('channel:demo')); //container.addMessageListener(listenerAdapter2, new PatternTopic('channel:demo2')); return container; } /** * 消息監(jiān)聽(tīng)器適配器，綁定消息處理器，利用反射技術(shù)調(diào)用消息處理器的業(yè)務(wù)方法 * @param demo 第一步的業(yè)務(wù)處理類(lèi) * @return */ @Bean MessageListenerAdapter listenerAdapter(Demo demo) { logger.info('----------------消息監(jiān)聽(tīng)器加載成功----------------'); // onMessage 就是方法名，基于反射調(diào)用 return new MessageListenerAdapter(demo, 'onMessage'); } /** * 多個(gè)交換機(jī)就多寫(xiě)一個(gè) * @param subCheckOrder * @return */ //@Bean //MessageListenerAdapter listenerAdapter2(SubCheckOrder subCheckOrder) { // logger.info('----------------消息監(jiān)聽(tīng)器加載成功----------------'); // return new MessageListenerAdapter(subCheckOrder, 'onMessage'); //} @Bean StringRedisTemplate template(RedisConnectionFactory connectionFactory) { return new StringRedisTemplate(connectionFactory); }}

4、消息發(fā)布

@Autowiredprivate RedisTemplate<String, Object> redisTemplate; redisTemplate.convertAndSend('channel:demo', '我是內(nèi)容');三、具體用法 socket連接成功。 socket消息推送時(shí)，把信息發(fā)布到redis中。socket服務(wù)訂閱redis的消息，訂閱成功后進(jìn)行推送。集群下的socket都能訂閱到消息，但是只有之前連接成功的節(jié)點(diǎn)能推送成功，其余的無(wú)法推送。