Springboot 整合 WebSocket ，使用STOMP协议+Redis 解决负载场景问题（二）

动我来个眼神

媒介


上一篇，简朴给各人整开了一下websocket，利用stomp方法。
那篇，便是考虑到单体的效劳利用websocket ，根据上一篇的整开，的确出成绩。
可是假如一旦是背载多台效劳的时分，那末便会呈现丧失成绩。

甚么？出有念过那个成绩？ 
没关系，看图教工具：

一向风格，我瞎绘了一张简图，大抵讲一下前后端利用websocket通信的局面。

 简析：
 后端某个效劳起了，整开了websocket做为 server，开放了一些节面endpoints ；
 前端效劳也起了，也整开了websocket做为 client，毗连server的websocket ；
 后端server 将每一个 前端client 毗连的 websocket session 皆存起去， 确保 明白谁是谁。
 如许正在server给client推收动静的时分，能包管推收，数据没有丧失。
websocket session map  是存正在于 后端效劳 的内乱存内里的 ，单一台后端 server，貌似出啥年夜成绩。

ok，我们简朴也理解了一下大抵的局面，接下去持续看图 ，多台websocket server 背载场景：

 简析：
如今做为统一个后端微效劳，整开websocket，起了两台。
如上图所示，假如 红色的client 跟 websocket挨交讲的时分，毗连的是上里的 浅蓝色websocket server；

浅蓝色websocket server 推收动静给  红色的client ，通信出成绩。
由于websocket server 确当天效劳session map内里寄存着 红色client的毗连websocket session；
动静丧失情况呈现：
我们背载了两台websocket server ，假如触收 websocket server 给红色client持续推收告诉动静， nginx/网闭 按照我们平常的背载均衡设置划定规矩，分收到了 绿色的 websocket server。
此时，绿色的 websocket server 确当天效劳session map内里 并出有  红色client的毗连websocket session ，以是会招致  告诉动静 丧失 。

处理计划：
既然成绩呈现了，那末我们便处理它，本篇便是介绍怎样经由过程 整开动静中心件来处理那个动静丧失的成绩。
我们采取的计划 是 将整开websocket 的 server效劳 （多台） 皆整开 redis做为动静中心件；
正在websocket  server 推收动静给websocket  client时， 先把动静拾到 redis内里 。
然后一切的websocket  server （不论几台效劳） 城市定阅此主题，获得到需求推收的数据，接下去 再推收给到对应的 destination 节面 （这时候候只要实正取当前client有毗连干系的 server扶效劳才气推收胜利，其他皆出有推收）。
做个简图：

或许许多人看到那内心里多几少有些迷惑， 实在底子缘故原由便是 多台 server 出法子同享 毗连session，假如能把session 连结起去同享，岂没有是处理了？
是的，思绪是对的，惋惜的是 websocekt session 是出有完成序列化接心的，没法利用相同redis来存储起去，然后反序列化获得。（可是实在能够经由过程redis存储相关websocket sessionkey 取节面的IP地点、端心，强止把恳求再次分收到准确的websocket server上里来。可是小我私家觉得这类方法没有是很好，以是本文仍是介绍redis的定阅/推收方法去处理那个成绩）
话未几道，进进代码环节。

注释


基于上一篇的根柢，开端魔改把redis参加出去。

 揭代码：

pom.xml ：

1.     <dependencies>
2.         <dependency>
3.             <groupId>org.springframework.boot</groupId>
4.             <artifactId>spring-boot-starter-data-redis</artifactId>
5.         </dependency>
6.         <dependency>
7.             <groupId>com.alibaba</groupId>
8.             <artifactId>fastjson</artifactId>
9.             <version>1.2.68</version>
10.         </dependency>
12.         <dependency>
13.             <groupId>org.springframework.boot</groupId>
14.             <artifactId>spring-boot-starter-websocket</artifactId>
15.         </dependency>
16.         <dependency>
17.             <groupId>org.springframework.boot</groupId>
18.             <artifactId>spring-boot-starter</artifactId>
19.         </dependency>
20.         <dependency>
21.             <groupId>org.springframework.boot</groupId>
22.             <artifactId>spring-boot-starter-test</artifactId>
23.             <scope>test</scope>
24.         </dependency>
25.     </dependencies>

复造代码

application.yml ：

1. server:
2.   port: 9908
3. spring:
4.   redis:
5.     host: 127.0.0.1
6.     port: 6379
7.     password: 123456

复造代码

RedisConfig.java ：
 

1. import com.fasterxml.jackson.annotation.JsonAutoDetect;
2. import com.fasterxml.jackson.annotation.PropertyAccessor;
3. import com.fasterxml.jackson.databind.ObjectMapper;
4. import com.stomp.stomptest.listener.RedisListener;
5. import org.springframework.cache.annotation.CachingConfigurerSupport;
6. import org.springframework.cache.annotation.EnableCaching;
7. import org.springframework.context.annotation.Bean;
8. import org.springframework.context.annotation.Configuration;
9. import org.springframework.context.annotation.Primary;
10. import org.springframework.data.redis.connection.RedisConnectionFactory;
11. import org.springframework.data.redis.core.RedisTemplate;
12. import org.springframework.data.redis.listener.PatternTopic;
13. import org.springframework.data.redis.listener.RedisMessageListenerContainer;
14. import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
15. import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
16. import org.springframework.data.redis.serializer.StringRedisSerializer;
18. /**
19. * @Author JCccc
20. * @Description   redis设置
21. * @Date 2021/6/30 8:53
22. */
23. @Configuration
24. @EnableCaching
25. public class RedisConfig extends CachingConfigurerSupport {
27.     @Bean
28.     @Primary
29.     public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
31.         RedisTemplate<String, Object> template = new RedisTemplate<>();
32.         template.setConnectionFactory(factory);
34.         Jackson2JsonRedisSerializer jacksonSeial = new Jackson2JsonRedisSerializer(Object.class);
35.         StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
36.         ObjectMapper om = new ObjectMapper();
37.         om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
38.         om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
39.         jacksonSeial.setObjectMapper(om);
40.         template.setValueSerializer(jacksonSeial);
41.         template.setKeySerializer(stringRedisSerializer);
42.         template.setHashKeySerializer(stringRedisSerializer);
43.         template.setHashValueSerializer(jacksonSeial);
44.         template.afterPropertiesSet();
45.         return template;
46.     }
48.     @Bean
49.     RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
50.                                             MessageListenerAdapter topicAdapter) {
51.         RedisMessageListenerContainer container = new RedisMessageListenerContainer();
52.         container.setConnectionFactory(connectionFactory);
53.         //定阅了主题 webSocketMsgPush
54.         container.addMessageListener(topicAdapter, new PatternTopic("webSocketMsgPush"));
55.         return container;
56.     }
57.     /**
58.      * 动静监听器适配器，绑定动静处置器
59.      *
60.      * @return
61.      */
62.     @Bean
63.     MessageListenerAdapter topicAdapter() {
64.         return new MessageListenerAdapter(new RedisListener());
65.     }
69. }

复造代码

简析：

 WebSocketConfig.java :

1. import org.springframework.context.annotation.Configuration;
2. import org.springframework.messaging.simp.config.MessageBrokerRegistry;
3. import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
4. import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
5. import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
6. import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
7. import org.springframework.web.socket.config.annotation.WebSocketTransportRegistration;
10. /**
11. * @Author JCccc
12. * @Description   EnableWebSocketMessageBroker-注解开启STOMP和谈去传输基于代办署理的动静，此时掌握器撑持利用@MessageMapping
13. * @Date 2021/6/30 8:53
14. */
15. @Configuration
16. @EnableWebSocketMessageBroker
17. public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
18.     private static long HEART_BEAT = 10000;
19.     @Override
20.     public void configureMessageBroker(MessageBrokerRegistry config) {
21.         ThreadPoolTaskScheduler te = new ThreadPoolTaskScheduler();
22.         te.setPoolSize(1);
23.         te.setThreadNamePrefix("wss-heartbeat-thread-");
24.         te.initialize();
25.         config.enableSimpleBroker("/user", "/topic").setHeartbeatValue(new long[]{HEART_BEAT, HEART_BEAT}).setTaskScheduler(te);
26.     }
28.     /**
29.      * 开放节面
30.      * @param registry
31.      */
32.     @Override
33.     public void registerStompEndpoints(StompEndpointRegistry registry) {
34.         //注册两个STOMP的endpoint，别离用于播送战面对面
35.         //播送
36.         registry.addEndpoint("/publicServer").withSockJS();
38.         //面对面
39.         registry.addEndpoint("/privateServer").withSockJS();
40.     }
41.     @Override
42.     public void configureWebSocketTransport(WebSocketTransportRegistration registration) {
43.         registration.setMessageSizeLimit(500 * 1024 * 1024);
44.         registration.setSendBufferSizeLimit(1024 * 1024 * 1024);
45.         registration.setSendTimeLimit(200000);
46.     }
49. }

复造代码

InjectServiceUtil.java :

1. import com.stomp.stomptest.producer.PushMessage;
2. import org.springframework.beans.factory.annotation.Autowired;
3. import org.springframework.stereotype.Component;
5. import javax.annotation.PostConstruct;
9. /**
10. * @Author JCccc
11. * @Description   pushMessage (单例)
12. * @Date 2021/6/30 8:53
13. */
14. @Component
15. public class InjectServiceUtil {
17.     @Autowired
18.     private PushMessage pushMessage;
20.     @PostConstruct
21.     public void init(){
22.         InjectServiceUtil.getInstance().pushMessage = this.pushMessage;
23.     }
25.     /**
26.      *  完成单例 start
27.      */
28.     private static class SingletonHolder {
29.         private static final InjectServiceUtil INSTANCE = new InjectServiceUtil();
30.     }
31.     private InjectServiceUtil (){}
32.     public static final InjectServiceUtil getInstance() {
33.         return SingletonHolder.INSTANCE;
34.     }
35.     /**
36.      *  完成单例 end
37.      */
38.     public PushMessage pushMessage(){
39.         return InjectServiceUtil.getInstance().pushMessage;
40.     }
41. }

复造代码

RedisListener.java :

1. import org.springframework.data.redis.connection.Message;
2. import org.springframework.data.redis.connection.MessageListener;
4. /**
5. * @Author JCccc
6. * @Description   redis监听动静
7. * @Date 2021/6/30 8:53
8. */
9. public class RedisListener implements MessageListener {
11.     @Override
12.     public void onMessage(Message message, byte[] bytes) {
13.         System.out.println("步调1.监听到需求停止背载转收的动静：" + message.toString());
14.         InjectServiceUtil.getInstance().pushMessage().send(message.toString());
15.     }
16. }

复造代码

简析：

Message.java ：

1. /**
2. * @Author JCccc
3. * @Description
4. * @Date 2021/8/20 9:26
5. */
6. public class Message {
8.     /**
9.      * 动静编码
10.      */
11.     private String code;
13.     /**
14.      * 去自（包管独一）
15.      */
16.     private String form;
18.     /**
19.      * 来自（包管独一）
20.      */
21.     private String to;
23.     /**
24.      * 内乱容
25.      */
26.     private String content;
28.     public String getCode() {
29.         return code;
30.     }
32.     public void setCode(String code) {
33.         this.code = code;
34.     }
36.     public String getForm() {
37.         return form;
38.     }
40.     public void setForm(String form) {
41.         this.form = form;
42.     }
44.     public String getTo() {
45.         return to;
46.     }
48.     public void setTo(String to) {
49.         this.to = to;
50.     }
52.     public String getContent() {
53.         return content;
54.     }
56.     public void setContent(String content) {
57.         this.content = content;
58.     }
59. }

复造代码

PushMessage.java  ：

1. import com.alibaba.fastjson.JSON;
2. import com.stomp.stomptest.pojo.Message;
3. import org.springframework.beans.factory.annotation.Autowired;
4. import org.springframework.messaging.simp.SimpMessagingTemplate;
5. import org.springframework.stereotype.Service;
7. /**
8. * @Author JCccc
9. * @Description   动静收收
10. * @Date 2021/6/30 8:53
11. */
12. @Service
13. public class PushMessage {
14.     @Autowired
15.     private SimpMessagingTemplate template;
17.     public void send(String msgJson){
19.         System.out.println("步调2.行将推收给websocket server：");
21.         Message message = JSON.parseObject(msgJson, Message.class);
23.         System.out.println("步调2.1 动静收给："+message.getTo());
24.         System.out.println("步调2.2 收收内乱容是："+message.getContent());
25.         template.convertAndSendToUser(message.getTo(), "/message", message.getContent());
26.         System.out.println("----------动静收收终了----------");
27.         
28.         //播送
29.        // template.convertAndSend("/topic/public",chatMessage);
30.     }
31. }

复造代码

最初是 TestController.java （测试接心。模仿触收体系推收动静）：

1. @Controller
2. public class TestController {
3.     @Autowired
4.     public SimpMessagingTemplate template;
6.     @ResponseBody
7.     @RequestMapping("/pushToOneTest")
8.     public String sendMessage(@RequestBody Message message) {
9.         String messageJson = JSON.toJSONString(message);
10.         System.out.println("！！！体系筹办做动静推收！！！");
11.         stringRedisTemplate.convertAndSend("webSocketMsgPush",messageJson);
12.         return "收收胜利";
13.     }
14. }

复造代码

ok，项目跑起去，挪用一下测试接心，看看全部流程：
能够看到动静一般推收：

看下代码内里挨印的疑息（各人初教的能够之间挨debug看下贱程会更好）：

能够看到，这类场景，不论背载了几台， 动静皆先到 redis内里来。

每台server皆经由过程redisListener 监听主题，获得到相关 动静 ， 然后才开端利用本地的websocket session 来找到各自效劳器内乱能否存正在当前用户的毗连数据，然后推收进来。
那些找没有到的没法举办推收进来，当然相形见绌而已，可是 我们的背载成绩处理了。
祝贺我们本人。


ok，该篇便到那。






最初另有便是之前的一篇，利用rabbitmq做为动静代理的：
《Springboot 整开Websocket+Stomp和谈+RabbitMQ做动静代理 真例教程》
https://blog.csdn.net/qq_35387940/article/details/108276136











免责声明：假如进犯了您的权益，请联络站少，我们会实时删除侵权内乱容，感谢协作！