开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> Springboot 整合 WebSocket ，使用STOMP协议+Redis 解决负载场景问题（二） -> 正文阅读

[大数据]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 ：

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.68</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

application.yml ：

server:
  port: 9908
spring:
  redis:
    host: 127.0.0.1
    port: 6379
    password: 123456

RedisConfig.java ：
?

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.stomp.stomptest.listener.RedisListener;
import org.springframework.cache.annotation.CachingConfigurerSupport;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

/**
 * @Author JCccc
 * @Description   redis配置
 * @Date 2021/6/30 8:53
 */
@Configuration
@EnableCaching
public class RedisConfig extends CachingConfigurerSupport {

    @Bean
    @Primary
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {

        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(factory);

        Jackson2JsonRedisSerializer jacksonSeial = new Jackson2JsonRedisSerializer(Object.class);
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jacksonSeial.setObjectMapper(om);
        template.setValueSerializer(jacksonSeial);
        template.setKeySerializer(stringRedisSerializer);
        template.setHashKeySerializer(stringRedisSerializer);
        template.setHashValueSerializer(jacksonSeial);
        template.afterPropertiesSet();
        return template;
    }

    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
                                            MessageListenerAdapter topicAdapter) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        //订阅了主题 webSocketMsgPush
        container.addMessageListener(topicAdapter, new PatternTopic("webSocketMsgPush"));
        return container;
    }
    /**
     * 消息监听器适配器，绑定消息处理器
     *
     * @return
     */
    @Bean
    MessageListenerAdapter topicAdapter() {
        return new MessageListenerAdapter(new RedisListener());
    }



}

简析：

?WebSocketConfig.java :

import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketTransportRegistration;


/**
 * @Author JCccc
 * @Description   EnableWebSocketMessageBroker-注解开启STOMP协议来传输基于代理的消息，此时控制器支持使用@MessageMapping
 * @Date 2021/6/30 8:53
 */
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
    private static long HEART_BEAT = 10000;
    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        ThreadPoolTaskScheduler te = new ThreadPoolTaskScheduler();
        te.setPoolSize(1);
        te.setThreadNamePrefix("wss-heartbeat-thread-");
        te.initialize();
        config.enableSimpleBroker("/user", "/topic").setHeartbeatValue(new long[]{HEART_BEAT, HEART_BEAT}).setTaskScheduler(te);
    }

    /**
     * 开放节点
     * @param registry
     */
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        //注册两个STOMP的endpoint，分别用于广播和点对点
        //广播
        registry.addEndpoint("/publicServer").withSockJS();

        //点对点
        registry.addEndpoint("/privateServer").withSockJS();
    }
    @Override
    public void configureWebSocketTransport(WebSocketTransportRegistration registration) {
        registration.setMessageSizeLimit(500 * 1024 * 1024);
        registration.setSendBufferSizeLimit(1024 * 1024 * 1024);
        registration.setSendTimeLimit(200000);
    }


}

InjectServiceUtil.java :

import com.stomp.stomptest.producer.PushMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;



/**
 * @Author JCccc
 * @Description   pushMessage (单例)
 * @Date 2021/6/30 8:53
 */
@Component
public class InjectServiceUtil {

    @Autowired
    private PushMessage pushMessage;

    @PostConstruct
    public void init(){
        InjectServiceUtil.getInstance().pushMessage = this.pushMessage;
    }

    /**
     *  实现单例 start
     */
    private static class SingletonHolder {
        private static final InjectServiceUtil INSTANCE = new InjectServiceUtil();
    }
    private InjectServiceUtil (){}
    public static final InjectServiceUtil getInstance() {
        return SingletonHolder.INSTANCE;
    }
    /**
     *  实现单例 end
     */
    public PushMessage pushMessage(){
        return InjectServiceUtil.getInstance().pushMessage;
    }
}

RedisListener.java :

import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;

/**
 * @Author JCccc
 * @Description   redis监听消息
 * @Date 2021/6/30 8:53
 */
public class RedisListener implements MessageListener {

    @Override
    public void onMessage(Message message, byte[] bytes) {
        System.out.println("步骤1.监听到需要进行负载转发的消息：" + message.toString());
        InjectServiceUtil.getInstance().pushMessage().send(message.toString());
    }
}

简析：

Message.java ：

/**
 * @Author JCccc
 * @Description
 * @Date 2021/8/20 9:26
 */
public class Message {

    /**
     * 消息编码
     */
    private String code;

    /**
     * 来自（保证唯一）
     */
    private String form;

    /**
     * 去自（保证唯一）
     */
    private String to;

    /**
     * 内容
     */
    private String content;

    public String getCode() {
        return code;
    }

    public void setCode(String code) {
        this.code = code;
    }

    public String getForm() {
        return form;
    }

    public void setForm(String form) {
        this.form = form;
    }

    public String getTo() {
        return to;
    }

    public void setTo(String to) {
        this.to = to;
    }

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }
}

PushMessage.java? ：

import com.alibaba.fastjson.JSON;
import com.stomp.stomptest.pojo.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Service;

/**
 * @Author JCccc
 * @Description   消息发送
 * @Date 2021/6/30 8:53
 */
@Service
public class PushMessage {
    @Autowired
    private SimpMessagingTemplate template;

    public void send(String msgJson){

        System.out.println("步骤2.即将推送给websocket server：");

        Message message = JSON.parseObject(msgJson, Message.class);

        System.out.println("步骤2.1 消息发给："+message.getTo());
        System.out.println("步骤2.2 发送内容是："+message.getContent());
        template.convertAndSendToUser(message.getTo(), "/message", message.getContent());
        System.out.println("----------消息发送完毕----------");
        
        //广播
       // template.convertAndSend("/topic/public",chatMessage);
    }
}

最后是 TestController.java （测试接口。模拟触发系统推送消息）：

@Controller
public class TestController {
    @Autowired
    public SimpMessagingTemplate template;

    @ResponseBody
    @RequestMapping("/pushToOneTest")
    public String sendMessage(@RequestBody Message message) {
        String messageJson = JSON.toJSONString(message);
        System.out.println("！！！系统准备做消息推送！！！");
        stringRedisTemplate.convertAndSend("webSocketMsgPush",messageJson);
        return "发送成功";
    }
}

ok，项目跑起来，调用一下测试接口，看看整个流程：

可以看到消息正常推送：