socke.io长连接实现推送、版本控制、实时活跃用户量统计

一、场景描述：

微服务搭建一个平台版本控制平台：实时检查每个应用及应用下各个平台端的在线用户，定点指定各平台下的软件更新。市面上的长连接工具很多，我这里选择socke.io,因为并发这些都是有专业的人测试过了重点是跨平台ios、Android、web等以及各种语言的支持都比较全面。

二、模型简介

三、基本实现

1、springboot集成socket.io长连接

1.1、pom引入依赖

        <!-- 长连接  -->
        <dependency>
            <groupId>com.corundumstudio.socketio</groupId>
            <artifactId>netty-socketio</artifactId>
            <version>1.7.19</version>
        </dependency>

 1.2、配置参数初始化

socketio:
  host: localhost
  port: 10091
  # 设置最大每帧处理数据的长度，防止他人利用大数据来攻击服务器
  maxFramePayloadLength: 1048576
  # 设置http交互最大内容长度
  maxHttpContentLength: 1048576
  # socket连接数大小（如只监听一个端口boss线程组为1即可）
  bossCount: 1
  workCount: 100
  allowCustomRequests: true
  # 协议升级超时时间（毫秒），默认10秒。HTTP握手升级为ws协议超时时间
  upgradeTimeout: 1000000
  # Ping消息超时时间（毫秒），默认60秒，这个时间间隔内没有接收到心跳消息就会发送超时事件
  pingTimeout: 6000000
  # Ping消息间隔（毫秒），默认25秒。客户端向服务器发送一条心跳消息间隔
  pingInterval: 25000

1.3、长连接启动类配置

import com.corundumstudio.socketio.SocketConfig;
import com.corundumstudio.socketio.SocketIONamespace;
import com.corundumstudio.socketio.SocketIOServer;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;

/**
 * 长连接启动配置类
 */
@Configuration
public class SocketIOConfig implements InitializingBean {

    @Resource
    private SocketIOHandler socketIOHandler;

    @Value("${socketio.host}")
    private String host;

    @Value("${socketio.port}")
    private Integer port;

    @Value("${socketio.bossCount}")
    private int bossCount;

    @Value("${socketio.workCount}")
    private int workCount;

    @Value("${socketio.allowCustomRequests}")
    private boolean allowCustomRequests;

    @Value("${socketio.upgradeTimeout}")
    private int upgradeTimeout;

    @Value("${socketio.pingTimeout}")
    private int pingTimeout;

    @Value("${socketio.pingInterval}")
    private int pingInterval;

    public SocketIOServer socketIOServer;

    @Override
    public void afterPropertiesSet() throws Exception {
        SocketConfig socketConfig = new SocketConfig();
        socketConfig.setReuseAddress(true);
        socketConfig.setTcpNoDelay(true);
        socketConfig.setSoLinger(0);

        com.corundumstudio.socketio.Configuration configuration = new com.corundumstudio.socketio.Configuration();
        configuration.setSocketConfig(socketConfig);
        // host在本地测试可以设置为localhost或者本机IP，在Linux服务器跑可换成服务器IP
        configuration.setHostname(host);
        configuration.setPort(port);
        // socket连接数大小（如只监听一个端口boss线程组为1即可）
        configuration.setBossThreads(bossCount);
        configuration.setWorkerThreads(workCount);
        configuration.setAllowCustomRequests(allowCustomRequests);
        // 协议升级超时时间（毫秒），默认10秒。HTTP握手升级为ws协议超时时间
        configuration.setUpgradeTimeout(upgradeTimeout);
        // Ping消息超时时间（毫秒），默认60秒，这个时间间隔内没有接收到心跳消息就会发送超时事件
        configuration.setPingTimeout(pingTimeout);
        // Ping消息间隔（毫秒），默认25秒。客户端向服务器发送一条心跳消息间隔
        configuration.setPingInterval(pingInterval);
        socketIOServer = new SocketIOServer(configuration);
        //添加事件监听器
        socketIOServer.addListeners(socketIOHandler);
        //这里以后是app应用表里边查询出来的 应用列表for循环注册命名空间
        SocketIONamespace namespace = socketIOServer.addNamespace("/chat");
        SocketIONamespace namespace1 = socketIOServer.addNamespace("/应用一的id");
        SocketIONamespace namespace2 = socketIOServer.addNamespace("/应用二的id");
        SocketIONamespace namespace3 = socketIOServer.addNamespace("/应用三的id");
        SocketIONamespace namespace4 = socketIOServer.addNamespace("/应用四的id");
        //启动SocketIOServer
        socketIOServer.start();
        System.out.println("SocketIO启动完毕");
    }
}

1.4、长连接事件注册类

import com.corundumstudio.socketio.HandshakeData;
import com.corundumstudio.socketio.SocketIOClient;
import com.corundumstudio.socketio.annotation.OnConnect;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

/**
 * 长连接事件监听
 */
@Component
@Slf4j
public class SocketIOHandler {
    @OnConnect
    public void onConnect(SocketIOClient client) {
        if (!ObjectUtils.isEmpty(client)) {
            HandshakeData handshakeData = client.getHandshakeData();
            //根据不同的平台id进行注册 如 Android、ios、java、web等
            String platformId= handshakeData.getSingleUrlParam("platformId");
            if (!StringUtils.isEmpty(roomId)) {
                client.joinRoom(platformId);
            }
        }
    }
}

1.5、向外提供用户统计接口(获取应用总用户数或者应用下的某个平台用户数)

import com.corundumstudio.socketio.BroadcastOperations;
import com.corundumstudio.socketio.SocketIOClient;
import com.corundumstudio.socketio.SocketIONamespace;
import com.corundumstudio.socketio.SocketIOServer;
import org.springframework.util.ObjectUtils;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.Collection;

@RestController
public class Contrler {
    @Resource
    private SocketIOConfig socketIOConfig;

    @GetMapping("/getAllUser")
    public String getAllUser(@RequestBody @Validated GetAllUserDO getAllUserDO) {
        SocketIOServer socketIOServer = socketIOConfig.socketIOServer;
        if (!ObjectUtils.isEmpty(socketIOServer)) {
            SocketIONamespace namespace = socketIOServer.getNamespace("/"+getAllUserDO.getAppId());
            if (!ObjectUtils.isEmpty(namespace)) {
                if (!ObjectUtils.isEmpty(getAllUserDO.getPlatformId())) {
                    BroadcastOperations roomOperations = namespace.getRoomOperations(getAllUserDO.getPlatformId() + "");
                    return roomOperations.getClients().size() + "";
                }
                Collection<SocketIOClient> allClients = namespace.getAllClients();
                return allClients.size() + "";
            }
        }
        return "0";
    }
}

2、JS客户端建立长连接（ios、Android等每个平台都有相关sdk几行代码都能集成）

<html>
<head>
    <meta charset="utf-8"/>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/2.1.0/socket.io.js"></script>
 <body>
     <h1>Netty-socketio Demo Chat</h1>
     <br/>
     <div id="console" class="well">
     </div>
     消息总数：<div id="msgnum">0</di>    
 </body>     
 <script type="text/javascript">
         var socket =  io.connect('http://localhost:10091/chat?platformId=666');
         socket.
         socket.on('connect', function() {
             output('<span class="connect-msg">Client has connected to the server!</span>');
         });
         socket.on('disconnect', function() {
             output('<span class="disconnect-msg">The client has disconnected!</span>');
         });
         function sendDisconnect() {
             socket.disconnect();
         } 
     </script>
</head>
</html>

四、客户端与服务器联动（具体的版本推送和数据交互略）