devbox
正经夜光杯
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

websocket之netty-socketio高性能实时消息推送框架

作者：正经夜光杯 | 2024-08-13 08:05:24

赞

踩

netty-socketio

简介

websocket协议的出现，打破http那种只能由客户端发起请求，服务器响应的单项数据传输局面，采用websocket之后，可以由客户端推送数据到服务端，也可以由服务器主动推送数据到客户端，从而同时进行双向通信，便于数据实时更新推送

netty-socketio便是为了方便我们进行websocket开发的框架，底层采用netty，提供了可靠高效的性能

netty-socketio的官方github地址：https://github.com/mrniko/netty-socketio

入门demo

咋们先来一个入门的websocket程序，带大家先简单了解下netty-socketio的使用

demo相关代码地址：https://gitee.com/xumengqq/work-websocketw.git

项目结构图

依赖

  1.         <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter</artifactId>
  4.         </dependency>
  5.         <dependency>
  6.             <groupId>com.corundumstudio.socketio</groupId>
  7.             <artifactId>netty-socketio</artifactId>
  8.             <version>1.7.19</version>
  9.         </dependency>
  10.         <dependency>
  11.             <groupId>com.alibaba</groupId>
  12.             <artifactId>fastjson</artifactId>
  13.             <version>1.2.73</version>
  14.         </dependency>

配置

  1. # 自定义socket服务端配置
  2. websocket:
  3.   server:
  4.     host: localhost
  5.     port: 9999

服务端代码实现

netty-socketio框架采用的数据传输格式是["key","data"]，这是一个数组，前面的key是信令，可以自定义，后面是传递的数据，理解为对"key"信令发送了一条"data"数据。也可以类比成对某个接口发送了一条数据。后面的代码编写思路也是监听这个key信令上传的数据，或者主动对这个key信令下发数据

实体类和常量类
  1. @Data
  2. public class Message {
  3.     private String userId;
  4.     private String room;
  5.     private String data;
  6. }
  7.  
  8. public class EventConstant {
  9.  
  10.     public static final String TEST_EVENT = "test";
  11.  
  12.     public static final String MESSAGE = "message";
  13.  
  14.     public static final String USER_ID = "userId";
  15.  
  16.     public static final String JOIN_ROOM = "joinRoom";
  17.     public static final String LEAVE_ROOM = "leaveRoom";
  18.  
  19.     /**
  20.      * 获取该用户所有加入的房间
  21.      */
  22.     public static final String GET_MY_ROOMS = "getMyRooms";
  23.  
  24.     /**
  25.      * 房间内广播消息
  26.      * */
  27.     public static final String ROOM_BROADCAST = "roomBroadcast";
  28.     /**
  29.      * 全局广播消息
  30.      * */
  31.     public static final String GLOBAL_BROADCAST = "globalBroadcast";
  32.  
  33.     public static final String SEND_USER_DATA = "sendUserData";
  34. }
socket配置类编写
  1. @Configuration
  2. @Log4j2
  3. public class WebsocketConfig {
  4.  
  5.     @Value("${websocket.server.host:localhost}")
  6.     private String host;
  7.     @Value("${websocket.server.port:9999}")
  8.     private Integer websocketPort;
  9.  
  10.     /**
  11.      * 设置socket service基本配置
  12.      * @return
  13.      */
  14.     @Bean
  15.     public SocketIOServer getSocketIOServer(){
  16.         com.corundumstudio.socketio.Configuration configuration = new com.corundumstudio.socketio.Configuration();
  17.         configuration.setPort(websocketPort);
  18.         configuration.setHostname(host);
  19.         configuration.getSocketConfig().setTcpNoDelay(true);
  20.         configuration.getSocketConfig().setReuseAddress(true);
  21.         return new SocketIOServer(configuration);
  22.     }
  23.  
  24.     /**
  25.      * 扫描socket相关的注解，例如@OnConnect、@DisConnect、@OnEvent
  26.      * @return
  27.      */
  28.     @Bean
  29.     public SpringAnnotationScanner springAnnotationScanner(){
  30.         return new SpringAnnotationScanner(getSocketIOServer());
  31.     }
  32. }

netty-socketio还提供了命名空间的概念，简单来说就是一个netty-socketio应用程序，里面有n个namespace，一个namespace里面又有n个room，用户默认会加入一个空字符""名称 的namespace和一个空字符""名称的room,下面有获取默认的namespace和room的代码和运行结果。绝大多数情况下使用默认的namespace就可以了，这里不做展开。

 

 其它的配置可以参考github上的说明：https://github.com/mrniko/netty-socketio/wiki/Configuration-details

  • setHostname If not set then bind address will be 0.0.0.0 or ::0

  • setPort The port the socket.io server will listen to

  • setJsonTypeFieldName defaults to "@class"

  • setJsonSupport Allows to setup custom implementation of JSON serialization/deserialization. See JsonSupport

  • setAuthorizationListener Authorization listener invoked on every handshake. Accepts all clients by default.

  • setStoreFactory Client store and pubsub factory, used to store session data and implements distributed pubsub. Defaults to MemoryStoreFactory, but RedisStoreFactory and HazelcastStoreFactory also implemented.

  • setPreferDirectBuffer (added in 1.5 version) Buffer allocation method used during packet encoding. Defaults to true.

  • setBossThreads (added in 1.5 version) boss-threads amount for netty

  • setWorkerThreads (added in 1.5 version) worker-threads amount for netty

  • setHeartbeatInterval Heartbeat interval (in seconds), defaults to 25

  • setHeartbeatTimeout Heartbeat timeout (in seconds), defaults to 60. Use 0 to disable it

  • setCloseTimeout Channel close timeout (in seconds) due to inactivity, defaults to 60

  • setContext Namespace, defaults to "/socket.io"

  • setAllowCustomRequests Allow to service custom requests that differ from socket.io protocol, defaults to false.

    • If true, add own handler which handle custom requests in order to avoid hang connections.

  • setPollingDuration Polling interval for XHR transport (in seconds), defaults to 20

  • setKeyStorePassword SSL key store password (for secure connections)

  • setKeyStore SSL key store stream, maybe appointed to any source

  • setMaxHttpContentLength Set maximum HTTP content length limit, defaults to 64KB.

    • If the length of the aggregated content exceeds this value, a TooLongFrameException will be raised.

  • setTransports Transports supported by server, defaults to [Transport.WEBSOCKET, Transport.FLASHSOCKET, Transport.XHRPOLLING]. Cannot be empty list

socket服务启动和销毁
  1. @SpringBootApplication
  2. @Log4j2
  3. public class SingleApplication implements CommandLineRunner, DisposableBean {
  4.  
  5.     @Autowired
  6.     private SocketIOServer socketIOServer;
  7.  
  8.     public static void main(String[] args) {
  9.         SpringApplication.run(SingleApplication.class, args);
  10.     }
  11.  
  12.     /**
  13.      * 启动socket服务器
  14.      * @param args
  15.      * @throws Exception
  16.      */
  17.     @Override
  18.     public void run(String... args) throws Exception {
  19.         log.info("socket start ...");
  20.         socketIOServer.start();
  21.         log.info("socket started");
  22.     }
  23.  
  24.     /**
  25.      * spring容器关闭时将socket服务也停止
  26.      * @throws Exception
  27.      */
  28.     @Override
  29.     public void destroy() throws Exception {
  30.         socketIOServer.stop();
  31.         log.info("socket server stop");
  32.     }
  33. }
客户端连接和断开
  1. @Component
  2. @Log4j2
  3. public class ConnectHandel {
  4.  
  5.     public static final String USER_ID = "userId";
  6.  
  7.     /**
  8.      * 客户端socket连接后触发
  9.      * @param client
  10.      */
  11.     @OnConnect
  12.     public void connect(SocketIOClient client){
  13.         // 获取登录信息，也可以在这里面进行权限判断
  14.         String userId = client.getHandshakeData().getSingleUrlParam(USER_ID);
  15.         log.info("connect userId {}, sessionUuid {}",userId,client.getSessionId());
  16.     }
  17.  
  18.     /**
  19.      * 客户端断开服务器时触发
  20.      * @param client
  21.      */
  22.     @OnDisconnect
  23.     public void disConnect(SocketIOClient client){
  24.         // 获取登录信息
  25.         String userId = client.getHandshakeData().getSingleUrlParam(USER_ID);
  26.         log.info("disConnect userId {}, sessionUuid {}",userId,client.getSessionId());
  27.     }
  28. }
业务代码
客户端和服务器简单交互
  1.     @Autowired
  2.     private SocketIOServer socketIOServer;
  3.  
  4.     @OnEvent(MESSAGE)
  5.     public void messageEvent(SocketIOClient client,String message) throws InterruptedException {
  6.         String userId = client.getHandshakeData().getSingleUrlParam(USER_ID);
  7.         log.info("testEvent userId {}, sessionUuid {}, message {}",userId,client.getSessionId(),message);
  8.  
  9.         // 业务处理,这里写一个模拟demo
  10.         // 向客户端发消息
  11.         client.sendEvent(MESSAGE,"服务处理中...");
  12.  
  13.         Thread.sleep(1000);
  14.         // 业务处理结束后再向客户端发消息
  15.         client.sendEvent(MESSAGE,"服务处理完成!");
  16.     }
加入房间
  1.     @OnEvent(JOIN_ROOM)
  2.     public void joinRoom(SocketIOClient client,String room){
  3.         client.joinRoom(room);
  4.     }
离开房间
  1.     @OnEvent(LEAVE_ROOM)
  2.     public void leaveRoom(SocketIOClient client,String room){
  3.         client.leaveRoom(room);
  4.     }
房间内广播消息
  1.     @OnEvent(ROOM_BROADCAST)
  2.     public void roomBroadcast(SocketIOClient client, String data){
  3.         Message message = JSON.parseObject(data, Message.class);
  4.         roomBroadcastData(MESSAGE,message.getRoom(),message.getData(),null);
  5.     }
  6.     public void roomBroadcastData(String eventKey,String room,Object data,SocketIOClient excludeClient){
  7.         if (Objects.isNull(excludeClient)){
  8.             socketIOServer.getRoomOperations(room).sendEvent(eventKey,data);
  9.         }else {
  10.             socketIOServer.getRoomOperations(room).sendEvent(eventKey,excludeClient,data);
  11.         }
  12.     }
全局广播消息
  1.     @OnEvent(GLOBAL_BROADCAST)
  2.     public void globalBroadcast(SocketIOClient client, String data){
  3.         globalBroadcastData(MESSAGE,data,null);
  4.     }
  5.     public void globalBroadcastData(String eventKey,Object data,SocketIOClient excludeClient){
  6.         if(Objects.isNull(excludeClient)){
  7.             socketIOServer.getBroadcastOperations().sendEvent(eventKey,data);
  8.         }else {
  9.             socketIOServer.getBroadcastOperations().sendEvent(eventKey,excludeClient,data);
  10.         }
  11.     }
 点对点消息发送

这个需要存储userId和client的对应关系，知道需要发送的目标userId，就能获取到client，然后直接发消息既可

我们在@OnConnect和@DisConnect中来绑定和解除userId和client的关系

客户端

客户端代码我是在网上找的，大家可按自己的方式实现既可

js方式
  1. <!DOCTYPE html>
  2. <html>
  3. <head>
  4.     <meta http-equiv="Content-Type" content="text/html; charset=utf-8"/>
  5.     <title>SocketIO客户端测试环境</title>
  6.     <base>
  7.     <script src="https://cdn.bootcss.com/jquery/3.3.1/jquery.min.js"></script>
  8.     <script src="https://cdn.bootcss.com/socket.io/2.1.1/socket.io.js"></script>
  9.     <style>
  10.         body {
  11.             padding: 20px;
  12.         }
  13.         #console {
  14.             height: 450px;
  15.             overflow: auto;
  16.         }
  17.         .connect-msg {
  18.             color: green;
  19.         }
  20.         .disconnect-msg {
  21.             color: red;
  22.         }
  23.     </style>
  24. </head>
  25.  
  26. <body>
  27. <h1>客户端测试环境</h1>
  28. <hr style="height:1px;border:none;border-top:1px solid black;" />
  29.  
  30. <div style="width: 700px; float: left">
  31. <h3>SocketClient建立连接</h3>
  32. <div style="border: 1px;">
  33.  
  34.     <label>socketio服务端地址:</label>
  35.     <!--
  37.       http://localhost 服务端ip
  38.                   9999 服务端socket端口（服务端提供）
  39.                 test或socketIO 命名空间（可自定义）如果不定义命名空间，默认是/     比如：http://localhost:9999?userId=12345
  40.                 userId 用户id参数（可自定义）
  42.        ps:因为我定义了命名空间/test和/socketIO，所以我这里也可以用
  43.           http://localhost:9999/test?userId=12345
  44.           http://localhost:9999/socketIO?userId=12345
  45.          这里我用http://localhost:9999?userId=12345建立连接，因为这里还不涉及到请求不同命名空间的方法
  46.       -->
  47.     <input type="text" id="url" value="ws://localhost:9999?userId=12345" style="width: 500px;">
  48.     <br>
  49.     <br>
  50.     <button id="connect" style="width: 100px;">建立连接</button>
  51.     <button id="disconnect" style="width: 100px;">断开连接</button>
  52.  
  53.  
  54. </div>
  55.  
  56. <hr style="height:1px;border:none;border-top:1px solid black;" />
  57.  
  58. <h3>SocketClient发送消息</h3>
  59. <div style="border: 1px;">
  60.     <label>socketEvent名称:</label><input type="text" id="socketEvent" value="getUserRooms">
  61.     <br><br>
  62.     <textarea  id="content" maxlength="1000" cols="40" rows="5" placeholder="请输入内容"></textarea>
  63.     <button id="send" style="width: 100px;">发送消息</button>
  64. </div>
  65.  
  66. <hr style="height:1px;border:none;border-top:1px solid black;" />
  67.  
  68. </div>
  69. <div style="float: left;margin-left: 50px;">
  70.     <h3>SocketIO互动消息</h3>
  71.     <button id="clean" style="width: 100px;">清理输出</button>
  72.     <div id="console" class="well"></div>
  73. </div>
  74.  
  75.  
  76. </body>
  77.  
  78. <script type="text/javascript">
  79.     var socket ;
  80.     var errorCount = 0;
  81.     var isConnected = false;
  82.     var maxError = 5;
  83.  
  84.     //连接
  85.     function connect(url) {
  86.  
  87.         //var opts = {
  88.         //    query: 'userId='+userId
  89.         //};
  90.         //socket = io.connect(url, opts);
  91.  
  92.         socket = io.connect(url);
  93.  
  94.         //socket.nsp = "/socketIO";//定义命名空间
  95.  
  96.         console.log(socket)
  97.  
  98.         //监听本次连接回调函数
  99.         socket.on('connect', function () {
  100.             isConnected =true;
  101.             console.log("连接成功");
  102.             serverOutput('<span class="connect-msg"><font color="blue">'+getNowTime()+'&nbsp;</font>连接成功</span>');
  103.             errorCount=0;
  104.         });
  105.         //监听消息
  106.         socket.on('message', function (data) {
  107.             output('<span class="connect-msg"><font color="blue">'+getNowTime()+'&nbsp;</font>' + data + ' </span>');
  108.             console.log(data);
  109.         });
  110.  
  111.         //监听断开
  112.         socket.on('disconnect', function () {
  113.             isConnected =false;
  114.             console.log("连接断开");
  115.             serverOutput('<span class="disconnect-msg"><font color="blue">'+getNowTime()+'&nbsp;</font>' + '已下线! </span>');
  116.         });
  117.         //监听断开错误
  118.         socket.on('connect_error', function(data){
  119.             serverOutput('<span class="disconnect-msg"><font color="blue">'+getNowTime()+'&nbsp;</font>;' + '连接错误-'+data+' </span>');
  120.             errorCount++;
  121.             if(errorCount>=maxError){
  122.                 socket.disconnect();
  123.             }
  124.         });
  125.         //监听连接超时
  126.         socket.on('connect_timeout', function(data){
  127.             serverOutput('<span class="disconnect-msg"><font color="blue">'+getNowTime()+'&nbsp;</font>' + '连接超时-'+data+' </span>');
  128.             errorCount++;
  129.             if(errorCount>=maxError){
  130.                 socket.disconnect();
  131.             }
  132.         });
  133.         //监听错误
  134.         socket.on('error', function(data){
  135.             serverOutput('<span class="disconnect-msg"><font color="blue">'+getNowTime()+'&nbsp;</font>' + '系统错误-'+data+' </span>');
  136.             errorCount++;
  137.             if(errorCount>=maxError){
  138.                 socket.disconnect();
  139.             }
  140.         });
  141.  
  142.         /*socket.on('ack', function(data){
  144.             console.log("ack:"+data)
  145.             var str = '消息发送失败';
  146.             if(data==1){
  147.                 str = '消息发送成功';
  148.             }
  149.             serverOutput('<span class="connect-msg"><font color="blue">'+getNowTime()+'&nbsp;</font>' + str+' </span>');
  151.         });*/
  152.     }
  153.  
  154.     function output(message) {
  155.         var element = $("<div>" + " " + message + "</div>");
  156.         $('#console').prepend(element);
  157.     }
  158.  
  159.     function serverOutput(message) {
  160.         var element = $("<div>" + message + "</div>");
  161.         $('#console').prepend(element);
  162.     }
  163.  
  164.     //连接
  165.     $("#connect").click(function(){
  166.         if(!isConnected){
  167.             var url =  $("#url").val();
  168.             connect(url);
  169.         }else {
  170.             serverOutput('<span class="disconnect-msg"><font color="blue">'+getNowTime()+'&nbsp;</font>' + '已经成功建立连接，不要重复建立！！！ </span>');
  171.         }
  172.     })
  173.  
  174.  
  175.     //断开连接
  176.     $("#disconnect").click(function(){
  177.         if(isConnected){
  178.             socket.disconnect();
  179.         }
  180.     })
  181.  
  182.     //发送消息
  183.     $("#send").click(function(){
  184.         var socketEvent =  $("#socketEvent").val();//自定义的事件名称
  185.         var content  = $("#content").val();//发送的内容
  186.         socket.emit(socketEvent,content,function(data1,data2){
  187.             console.log("ack1:"+data1);
  188.             console.log("ack2:"+data2);
  189.         });
  190.     })
  191.  
  192.     //清理消息
  193.     $("#clean").click(function(){
  194.         $('#console').html("");
  195.     })
  196.  
  197.     function getNowTime(){
  198.         var date=new Date();
  199.         var year=date.getFullYear(); //获取当前年份
  200.         var mon=date.getMonth()+1; //获取当前月份
  201.         var da=date.getDate(); //获取当前日
  202.         var h=date.getHours(); //获取小时
  203.         var m=date.getMinutes(); //获取分钟
  204.         var s=date.getSeconds(); //获取秒
  205.         var ms=date.getMilliseconds();
  206.         var d=document.getElementById('Date');
  207.         var date =year+'/'+mon+'/'+da+' '+h+':'+m+':'+s+':'+ms;
  208.         return date;
  209.     }
  210. </script>
  211. </html>
  212.
java 实现客户端
  1. package com.corundumstudio.socketio.demo.client;
  2.  
  3. import io.socket.client.IO;
  4. import io.socket.client.Socket;
  5. import io.socket.emitter.Emitter;
  6.  
  7. import java.net.URISyntaxException;
  8.  
  9.  
  10. public class DemoSocketClient {
  11.  
  12.     public static void main(String[] args) throws URISyntaxException, InterruptedException {
  13.         IO.Options options = new IO.Options();
  14.         options.transports = new String[]{"websocket"};
  15.         options.reconnectionAttempts = 2;
  16.         options.reconnectionDelay = 1000;//失败重连的时间间隔
  17.         options.timeout = 500;//连接超时时间(ms)
  18.  
  19.         final Socket socket = IO.socket("http://localhost:9999/?userId=12345", options);
  20.  
  21.         socket.on(Socket.EVENT_CONNECT, new Emitter.Listener() {
  22.             @Override
  23.             public void call(Object... args) {
  24.                 socket.send("hello");
  25.             }
  26.         });
  27.  
  28.         socket.on(Socket.EVENT_DISCONNECT, new Emitter.Listener() {
  29.             @Override
  30.             public void call(Object... args) {
  31.                 System.out.println("连接关闭");
  32.             }
  33.         });
  34.  
  35.         socket.on(Socket.EVENT_MESSAGE, new Emitter.Listener() {
  36.             @Override
  37.             public void call(Object... args) {
  38.                 System.out.println("sessionId:" + socket.id());
  39.                 for (Object obj : args) {
  40.                     System.out.println(obj);
  41.                 }
  42.                 System.out.println("收到服务器应答，将要断开连接...");
  43.                 socket.disconnect();
  44.             }
  45.         });
  46.         socket.connect();
  47.     }
  48. }

 

生产集群方案

上面的demo是单机版本，但是，生产上服务都是集群存在的，至少也要两台，双节点防止单点故障；可是websocket服务搭建集群会有个问题，就是服务间通信的问题；比如，Aclient连接上Aserver，Bclient连接上了Bserver，这时候，Aclient就没法对Bclient进行发送消息和接受 消息。

这里提供几种解决方案

Redis存储方案

redis是官方推荐的解决方案，这个在Github上可以看到。配置之后，利用redis这个中介，可以通过广播的方式对其他客户端进行通信，底层是用了redis的发布订阅功能。有两种通信方式，可以通过全局广播通知到其他客户端，也可以对加入同一个房间内的客户端进行广播。但是这个有局限性，netty-socketio只提供了广播的redis发布订阅方案，点对点方式并没有提供。

pom依赖
  1. <!--        添加redisson依赖-->
  2.         <dependency>
  3.             <groupId>org.redisson</groupId>
  4.             <artifactId>redisson-spring-boot-starter</artifactId>
  5.             <version>3.15.4</version>
  6.         </dependency>
yaml配置
  1. # redis
  2. spring:
  3.   redis:
  4.     host: 121.196.111.111  #地址
  5.     port: 6379      #端口号
  6.     database: 0     #0号库
  7.     password: password     #密码
配置类
  1. // redisson的配置类
  2. @Configuration
  3. public class RedissonConfig {
  4.     @Value("${spring.redis.host}")
  5.     private String host;
  6.  
  7.     @Value("${spring.redis.password}")
  8.     private String password;
  9.  
  10.     @Value("${spring.redis.port}")
  11.     private Integer port;
  12.  
  13.     @Value("${spring.redis.database:0}")
  14.     private Integer database;
  15.  
  16.     @Value("${spring.redis.timeout:3000}")
  17.     private Integer timeout;
  18.  
  19.     @Bean
  20.     public RedissonClient redissonClient(){
  21.         Config config = new Config();
  22. //        config.useClusterServers().addNodeAddress("redis://").setPassword("");
  23.         config.useSingleServer().setAddress("redis://"+ host +":" + port).setPassword(password).setDatabase(database).setTimeout(timeout);
  24.         RedissonClient redissonClient = Redisson.create(config);
  25.         return redissonClient;
  26.     }
  27. }
  28.  
  29.  
  30. // 在SocketIOServer配置上redisson既可使用
  31.     @Autowired
  32.     private RedissonClient redissonClient;
  33.     @Bean
  34.     public SocketIOServer getSocketIOServer(){
  35.         com.corundumstudio.socketio.Configuration configuration = new com.corundumstudio.socketio.Configuration();
  36.         configuration.setPort(websocketPort);
  37.         configuration.setHostname(host);
  38.         configuration.getSocketConfig().setTcpNoDelay(true);
  39.         configuration.getSocketConfig().setReuseAddress(true);
  40.         // 设置redis存储方式
  41.         configuration.setStoreFactory(new RedissonStoreFactory(redissonClient));
  42.         return new SocketIOServer(configuration);
  43.     }
全局广播和房间内广播
  1.     @OnEvent(ROOM_BROADCAST)
  2.     public void roomBroadcast(SocketIOClient client, String data){
  3.         Message message = JSON.parseObject(data, Message.class);
  4.         roomBroadcastData(MESSAGE,message.getRoom(),message.getData(),null);
  5.     }
  6.  
  7.     @OnEvent(GLOBAL_BROADCAST)
  8.     public void globalBroadcast(SocketIOClient client, String data){
  9.         globalBroadcastData(MESSAGE,data,null);
  10.     }
  11.  
  12.     public void globalBroadcastData(String eventKey,Object data,SocketIOClient excludeClient){
  13.         if(Objects.isNull(excludeClient)){
  14.             socketIOServer.getBroadcastOperations().sendEvent(eventKey,data);
  15.         }else {
  16.             socketIOServer.getBroadcastOperations().sendEvent(eventKey,excludeClient,data);
  17.         }
  18.     }
  19.  
  20.     public void roomBroadcastData(String eventKey,String room,Object data,SocketIOClient excludeClient){
  21.         if (Objects.isNull(excludeClient)){
  22.             socketIOServer.getRoomOperations(room).sendEvent(eventKey,data);
  23.         }else {
  24.             socketIOServer.getRoomOperations(room).sendEvent(eventKey,excludeClient,data);
  25.         }
  26.     }
生产灰度发布

nginx重新加载配置会影响原有的websocket连接吗？可不可以通过更新路由配置实现websocket服务的灰度发布？

经过测试，在更改nginx配置信息后，比如将其中一个服务设置成down，输入nginx  -s reload命令，原有的websocket连接不会受到影响，但是新的连接不会再进入设置为down的服务，随时间推移，连接数就会越来越少，直至为0。因此，我们可以将需要发布的服务设置为down，然后再编写一个获取此服务连接client的数量的接口，当数量为0的时候，代表没有websocket连接了，此时，既可对这台服务器进行发布了。

获取某台服务器连接客户端数量

  1.     @Autowired
  2.     private SocketIOServer socketIOServer;
  3.  
  4.     @GetMapping("/getClientCount")
  5.     public Object getClientCount(){
  6.         Collection<SocketIOClient> allClients = socketIOServer.getAllClients();
  7.         int size = allClients.size();
  8.         return size;
  9.     }

nginx配置websocket服务路由

这里需要采用hash的路由方式；不知道是框架还是其他的什么原因，客户端会发起三次的普通http请求和一次http升级成websocket的请求，采用其他的路由方式会导致这些请求不到同一个服务器从而导致websocket连接报错。这需要注意一下

  1. worker_processes  1;
  2.  
  3. events {
  4.     worker_connections  1024;
  5. }
  6.  
  7. http {
  8. 	map $http_upgrade $connection_upgrade {
  9. 		default upgrade;
  10. 		'' close;
  11. 	}
  12.  
  13.     upstream websocket {
  14.     	hash $remote_addr consistent;
  15.         server 127.0.0.1:9999 down;
  16.         server 127.0.0.1:9998;
  17.     }
  18.  
  19.     server {
  20.         listen 80;
  21.         location / {
  22.             proxy_pass http://websocket;
  23. 			proxy_http_version 1.1;
  24. 			proxy_set_header Upgrade $http_upgrade;
  25. 			proxy_set_header Connection "$connection_upgrade";
  26.             #proxy_set_header Host $host;
  27.         }
  28.     }
  29. }

Redis存储-点对点完善

之前的redis方案是netty-websocketio官方提供的集群解决方案，但此方案平没有解决点对点信息传输的问题，我们现在在此基础上还是通过redis的发布订阅来完善点对点信息推送。

当然，你也可以通过mq的方式来完成，个人觉得用redis的话，中小型项目就足够了，成本也低

思路：用户连上服务器，将用户和用户所连接的机器信息（这里指ip和端口号）保存在redis，断开连接则将redis上的记录清除掉；每台服务器订阅当前ip和端口号的topic，知道用户连接到那台服务器之后，发布消息则是发布到这个用户对应服务器ip和端口号的topic，这样，经过redis的订阅和发布，既可完成点对点的通信。（多台服务器分配多个topic方案是为了优化。如果采用一个topic，所有的机器都订阅同一个topic，然后每台机器再去赛选要发送的目标客户端是否连接在此机器上，这种方式会让其他非目标机器都进行一次赛选处理，略微有些影响性能，照成服务器不必要的开销）

ClientHub类

在之前代码的基础上，编写一个clientHub类，用于存储用户连接之前的信息

  1. @Component
  2. @Log4j2
  3. public class ClientHub implements CommandLineRunner {
  4.  
  5.     @Autowired
  6.     private SocketIOServer socketIOServer;
  7.  
  8.     @Autowired
  9.     private RedissonClient redissonClient;
  10.  
  11.     public static final String REDIS_TOPIC = "redis_topic";
  12.  
  13.     public static final String ONLINE_USER = "online_user_";
  14.  
  15.     /**
  16.      * 本机存储userId与客户端之间的连接关系
  17.      */
  18.     private static Map<String, SocketIOClient> userIdToClient = new HashMap<>();
  19.  
  20.     /**
  21.      * 集群模式下，点对点发送，一律采取这个方法，不能直接通过SocketClient发送
  22.      * @param fromUserId
  23.      * @param targetUserId
  24.      * @param key
  25.      * @param data
  26.      */
  27.     public void sendMessageByUserId(String fromUserId,String targetUserId,String key,String data){
  28.         Message message = new Message();
  29.         message.setFromUserId(fromUserId);
  30.         message.setTargetUserId(targetUserId);
  31.         message.setKey(key);
  32.         message.setData(data);
  33.  
  34.         SocketIOClient socketIOClient = userIdToClient.get(targetUserId);
  35.         // 如果是本机直连的，直接发送，不是则通过redis发布对应的机器订阅
  36.         if(!Objects.isNull(socketIOClient)){
  37.             socketIOClient.sendEvent(message.getKey(),message);
  38.         }else {
  39.             RBucket<String> bucket = redissonClient.getBucket(ONLINE_USER + targetUserId);
  40.             if(bucket.isExists()){
  41.                 String topic = bucket.get();
  42.                 this.pubMessage(topic,message);
  43.             }
  44.         }
  45.     }
  46.  
  47.     /**
  48.      * 用户连接时调用，存储用户连接之间的信息
  49.      * @param userId
  50.      * @param client
  51.      */
  52.     public void addUser(String userId,SocketIOClient client){
  53.         userIdToClient.put(userId,client);
  54.         // 存储在redis上
  55.         RBucket<String> bucket = redissonClient.getBucket(ONLINE_USER + userId);
  56.         bucket.set(getRedisTopic());
  57.     }
  58.  
  59.     /**
  60.      * 用户断开连接时调用，清楚用户和客户端之前的关系
  61.      * @param userId
  62.      */
  63.     public void deleteUser(String userId){
  64.         userIdToClient.remove(userId);
  65.         // 从redis上删除掉
  66.         RBucket<String> bucket = redissonClient.getBucket(ONLINE_USER + userId);
  67.         bucket.delete();
  68.     }
  69.  
  70.     /**
  71.      * 获取本socket服务的ip和端口号
  72.      * @return
  73.      */
  74.     private String getHostAndPort(){
  75.         String hostAddress = "";
  76.         try {
  77.             hostAddress = InetAddress.getLocalHost().getHostAddress();
  78.         } catch (UnknownHostException e) {
  79.             throw new RuntimeException(e);
  80.         }
  81.         int port = socketIOServer.getConfiguration().getPort();
  82.         return hostAddress+"_"+port;
  83.     }
  84.  
  85.     /**
  86.      * 获取这台机器redis订阅的topic
  87.      * @return
  88.      */
  89.     private String getRedisTopic(){
  90.         String hostAndPort = getHostAndPort();
  91.         return REDIS_TOPIC + "_" + hostAndPort;
  92.     }
  93.  
  94.     /**
  95.      * redis发布消息
  96.      * @param topic
  97.      * @param message
  98.      */
  99.     private void pubMessage(String topic, Message message){
  100.         redissonClient.getTopic(topic).publish(message);
  101.         log.info("pubMessage topic {} data {}",topic,JSON.toJSONString(message));
  102.     }
  103.  
  104.     /**
  105.      * redis订阅消息
  106.      * @param topic
  107.      */
  108.     public void subMessage(String topic){
  109.         redissonClient.getTopic(topic).addListener(Message.class,(channel,message)->{
  110.             log.info("topic {} submessage {}",topic, JSON.toJSONString(message));
  111.             String userId = message.getTargetUserId();
  112.             SocketIOClient socketIOClient = userIdToClient.get(userId);
  113.             if(Objects.nonNull(socketIOClient)){
  114.                 socketIOClient.sendEvent(message.getKey(),JSON.toJSONString(message));
  115.             }
  116.         });
  117.     }
  118.  
  119.     /**
  120.      * 服务器 对应的组件加载完成后，需要开启redis订阅功能
  121.      * @param args
  122.      * @throws Exception
  123.      */
  124.     @Override
  125.     public void run(String... args) throws Exception {
  126.         String redisTopic = getRedisTopic();
  127.         subMessage(redisTopic);
  128.         log.info("topic {} 已订阅",redisTopic);
  129.     }
  130. }
连接记录

建立连接和断开连接分别进行记录和清除工作

  1.     @Autowired
  2.     private ClientHub clientHub;
  3.  
  4.     /**
  5.      * 客户端socket连接后触发
  6.      * @param client
  7.      */
  8.     @OnConnect
  9.     public void connect(SocketIOClient client){
  10.         // 获取登录信息，也可以在这里面进行权限判断
  11.         String userId = client.getHandshakeData().getSingleUrlParam(USER_ID);
  12.         log.info("connect userId {}, sessionUuid {}",userId,client.getSessionId());
  13.         String url = client.getHandshakeData().getUrl();
  14.         log.info("url :{}",url);
  15.  
  16.         // 保存userId和client的对应关系
  17.         clientHub.addUser(userId,client);
  18.     }
  19.  
  20.     /**
  21.      * 客户端断开服务器时触发
  22.      * @param client
  23.      */
  24.     @OnDisconnect
  25.     public void disConnect(SocketIOClient client){
  26.         // 获取登录信息
  27.         String userId = client.getHandshakeData().getSingleUrlParam(USER_ID);
  28.         log.info("disConnect userId {}, sessionUuid {}",userId,client.getSessionId());
  29.  
  30.         // 解除userId和client的对应关系
  31.         clientHub.deleteUser(userId);
  32.     }
点对点发送消息业务使用
  1.      /**
  2.      * 点对点发送消息
  3.      * @param client
  4.      * @param data
  5.      * 注意：如果接受的参数是一个对象，那么前端页面发送数据的时候也必须是一个对象，不能用json字符串代替
  6.      * example：js代码 socket.emit('sendUserData', {data: "你好啊", targetUserId: "2233"});
  7.      */
  8.     @OnEvent(SEND_USER_DATA)
  9.     public void sendUserData(SocketIOClient client, AckRequest ackRequest,Message data){
  10.         // 获取登录信息
  11.         String userId = client.getHandshakeData().getSingleUrlParam(USER_ID);
  12.         log.info("sendUserData {}",JSON.toJSONString(data));
  13.         clientHub.sendMessageByUserId(userId,data.getTargetUserId(),MESSAGE,data.getData());
  14.     }

 这里只是简单的编写集群的思想，实际情况还有一些问题需要注意和解决，比如当某台机器挂了，这台机器连接的用户数据还是在redis上，其它机器就以为这些用户还是在线，但其实状态不对。这个就需要进行额外的处理了，比如存redis的时候设置过期时间，然后通过定时任务不断在redis上刷新在线用户信息等。

特殊场景集群方案

之前的redis方式，是通用的集群解决方案，基本上大部分的场景都适用，但有些时候业务场景比较特殊，也可以采取特殊的方式解决

特殊场景

用户之前的通信只在同一个房间内，比如视频会议，上课直播，游戏房间之类的，用户之间只和在同一个房间内的人进行通信，不需要对跨房间或者对房间外的用户进行实时消息推送。

思路

将房间看做一个单独的整体，只能分配在某一台服务器上，只要是这个房间的人需要连接服务器，就先获取该房间的服务器地址，客户端直接连上对应服务器，再进入房间，那么，该房间内的所有人都是在同一台服务器上，从而避免了跨服务器之间的通信场景，就相当于单机版。不过这个得手动控制房间的负载均衡，每台websocket服务器的访问路径也要自己控制好，生产的灰度发布也需要自行写代码控制。

spring事件模型

简述事件模型

spring提供的事件模型，其实也相当于一个发布订阅的工具，只不过是在单个程序内进行的，这里可以借助spring的事件，配合上多线程异步，可以更方便地进行websocket功能编码开发。从而达到，主要业务逻辑消息迅速实时推送，附加业务逻辑并发执行。

spring事件是spring自带的功能，不需要再额外引入依赖

定义事件

直接继承抽象类ApplicationEvent，然后自定义属性，自定义的属性用于在事件传播过程中记录自己需要的数据

  1. public class ConnectEvent extends ApplicationEvent {
  2.  
  3.     private SocketIOClient socketIOClient;
  4.  
  5.     public ConnectEvent(Object source,SocketIOClient socketIOClient) {
  6.         super(source);
  7.         this.socketIOClient = socketIOClient;
  8.     }
  9.  
  10.     public SocketIOClient getSocketIOClient() {
  11.         return socketIOClient;
  12.     }
  13. }
事件发布

通过ApplicationEventPublisher的实例既可发布自定义的事件

  1.     @Autowired
  2.     private ApplicationEventPublisher applicationEventPublisher;
  3.  
  4.     @OnConnect
  5.     public void connect(SocketIOClient client){
  6.         // 获取登录信息，也可以在这里面进行权限判断
  7.         String userId = client.getHandshakeData().getSingleUrlParam(USER_ID);
  8.         log.info("connect userId {}, sessionUuid {}",userId,client.getSessionId());
  9.         String url = client.getHandshakeData().getUrl();
  10.         log.info("url :{}",url);
  11.         // 也可以在这一步进行权限鉴定，鉴定不通过就发给客户端一个权限鉴定失败的消息，然后再断开连接
  12. //        client.sendEvent(MESSAGE,"权限不足");
  13. //        client.disconnect();
  14.  
  15.         // 保存userId和client的对应关系
  16.         clientHub.addUser(userId,client);
  17.  
  18.         // 发布一个用户连接事件
  19.         applicationEventPublisher.publishEvent(new ConnectEvent(this,client));
  20.     }
事件监听（订阅）

事件监听有两种书写方式，一种是直接实现ApplicationListener接口，泛型里面放入自定义事件

  1. @Component
  2. @Log4j2
  3. public class UserLoginListenerDemo implements ApplicationListener<ConnectEvent> {
  4.     @Override
  5.     public void onApplicationEvent(ConnectEvent event) {
  6.         SocketIOClient socketIOClient = event.getSocketIOClient();
  7.         String userId = socketIOClient.getHandshakeData().getSingleUrlParam(USER_ID);
  8.         {
  9.             // 业务代码。。。
  10.             log.info("UserLoginListenerDemo userId {}",userId);
  11.         }
  12.     }
  13. }

 第二中有点类似于controller的方式，在方法上用@EventListener注解来标识事件监听，在方法的参数中传入自定义的事件，并且如果有顺序需求的话，可以通过@Order来注明那个方法先调用（值小的先调用）

  1. @Component
  2. @Log4j2
  3. public class UserLoginListener {
  4.  
  5.     @EventListener
  6.     @Order(3)
  7.     public void connectListenerLog(ConnectEvent connectEvent){
  8.         SocketIOClient socketIOClient = connectEvent.getSocketIOClient();
  9.         String userId = socketIOClient.getHandshakeData().getSingleUrlParam(USER_ID);
  10.         {
  11.             log.info("user {} 登录的日志操作",userId);
  12.         }
  13.     }
  14.  
  15.     @EventListener
  16.     @Order(2)
  17.     public void connectListenerStore(ConnectEvent connectEvent){
  18.         SocketIOClient socketIOClient = connectEvent.getSocketIOClient();
  19.         String userId = socketIOClient.getHandshakeData().getSingleUrlParam(USER_ID);
  20.         {
  21.             log.info("user {} 登录的行为存表",userId);
  22.         }
  23.     }
  24.  
  25.     @EventListener
  26.     @Order(1)
  27.     public void connectListenerState(ConnectEvent connectEvent){
  28.         SocketIOClient socketIOClient = connectEvent.getSocketIOClient();
  29.         String userId = socketIOClient.getHandshakeData().getSingleUrlParam(USER_ID);
  30.         {
  31.             log.info("user {} 登录的状态更改",userId);
  32.         }
  33.     }
引入多线程异步处理

 上面的事件监听的方式是单线程的，如果仅仅是这样，感觉还不如直接定义方法进行调用，事件模型到显得有点多余。我们可以引入线程池和@Async，通过多线程并行处理事件，实现真正的发布订阅

创建一个spring提供的线程池

  1. @Configuration
  2. public class ThreadPoolConfig {
  3.  
  4.     @Bean
  5.     public ThreadPoolTaskExecutor autoLoginExecutor() {
  6.         ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
  7.         threadPoolTaskExecutor.setCorePoolSize(20);
  8.         threadPoolTaskExecutor.setMaxPoolSize(100);
  9.         threadPoolTaskExecutor.setQueueCapacity(200);
  10.         threadPoolTaskExecutor.setKeepAliveSeconds(60);
  11.         threadPoolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true);
  12.         threadPoolTaskExecutor.setAwaitTerminationSeconds(60);
  13.         threadPoolTaskExecutor.setThreadNamePrefix("event-task-");
  14.         threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardPolicy());
  15.         threadPoolTaskExecutor.initialize();
  16.         return threadPoolTaskExecutor;
  17.     }
  18. }

启动类上加上@EnableAsync注解

  1. @SpringBootApplication
  2. @Log4j2
  3. @EnableAsync
  4. public class ClusterEventApplication implements CommandLineRunner, DisposableBean {

 event监听方法上加上@Async表示异步执行

  1. @Component
  2. @Log4j2
  3. public class UserLoginListener {
  4.  
  5.     @EventListener
  6. //    @Order(3)
  7.     @Async
  8.     public void connectListenerLog(ConnectEvent connectEvent){
  9.         SocketIOClient socketIOClient = connectEvent.getSocketIOClient();
  10.         String userId = socketIOClient.getHandshakeData().getSingleUrlParam(USER_ID);
  11.         {
  12.             log.info("user {} 登录的日志操作",userId);
  13.         }
  14.     }
  15.  
  16.     @EventListener
  17. //    @Order(2)
  18.     @Async
  19.     public void connectListenerStore(ConnectEvent connectEvent){
  20.         SocketIOClient socketIOClient = connectEvent.getSocketIOClient();
  21.         String userId = socketIOClient.getHandshakeData().getSingleUrlParam(USER_ID);
  22.         {
  23.             log.info("user {} 登录的行为存表",userId);
  24.         }
  25.     }
  26.  
  27.     @EventListener
  28. //    @Order(1)
  29.     @Async
  30.     public void connectListenerState(ConnectEvent connectEvent){
  31.         SocketIOClient socketIOClient = connectEvent.getSocketIOClient();
  32.         String userId = socketIOClient.getHandshakeData().getSingleUrlParam(USER_ID);
  33.         {
  34.             log.info("user {} 登录的状态更改",userId);
  35.         }
  36.     }
  37. }

效果截图：

 

本文内容由网友自发贡献，转载请注明出处：https://www.wpsshop.cn/w/正经夜光杯/article/detail/973913
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号