springboot集成socketIo，实现webSocket_springboot集成socket.io

之前都是使用http接口请求，最近有用到WebSocket的场景，客户端之间长连接通讯，记录一下spring boot整合socketIO的过程。

1. 国际惯例，引入依赖
   
2. application.yml（或者是application.properties）配置文件添加socketIo配置
   
3. 添加socketIoConfig配置类

@Configuration
public class SocketConfig {
    private static final Logger logger = LoggerFactory.getLogger(SocketConfig.class);
    @Value("${socketIo.host}")
    private String host;
    @Value("${socketIo.port}")
    private Integer port;
    @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;
    @Value("${socketIo.maxFramePayloadLength}")
    private int maxFramePayloadLength;
    @Value("${socketIo.maxHttpContentLength}")
    private int maxHttpContentLength;
    
    /**SocketIOServer配置*/
    @Bean("socketIOServer")
    public SocketIOServer socketIOServer() {
        com.corundumstudio.socketio.Configuration config = new com.corundumstudio.socketio.Configuration();
        //配置host
        config.setHostname(host);
        //配置端口
        config.setPort(port);
        //开启Socket端口复用
        com.corundumstudio.socketio.SocketConfig socketConfig = new com.corundumstudio.socketio.SocketConfig();
        socketConfig.setReuseAddress(true);
        config.setSocketConfig(socketConfig);
        //连接数大小
        config.setWorkerThreads(workCount);
        //允许客户请求
        config.setAllowCustomRequests(allowCustomRequests);
        //协议升级超时时间(毫秒)，默认10秒，HTTP握手升级为ws协议超时时间
        config.setUpgradeTimeout(upgradeTimeout);
        //Ping消息超时时间(毫秒)，默认60秒，这个时间间隔内没有接收到心跳消息就会发送超时事件
        config.setPingTimeout(pingTimeout);
        //Ping消息间隔(毫秒)，默认25秒。客户端向服务器发送一条心跳消息间隔
        config.setPingInterval(pingInterval);
        //设置HTTP交互最大内容长度        config.setMaxHttpContentLength(maxHttpContentLength);
        //设置最大每帧处理数据的长度，防止他人利用大数据来攻击服务器       config.setMaxFramePayloadLength(maxFramePayloadLength);
        config.setTransports(Transport.POLLING, Transport.WEBSOCKET);
        /**config.setOrigin("*");*/
        return new SocketIOServer(config);
    }
	/**开启SocketIOServer注解支持*/
    @Bean
    public SpringAnnotationScanner springAnnotationScanner(SocketIOServer socketServer) {
        return new SpringAnnotationScanner(socketServer);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

4. 添加消息主体实体

@Data
@ToString
public class MessageDto extends BaseEntity{
    /**源客户端用户名*/
    private String sourceUserName;

    /**目标客户端用户名-入参*/
    private String targetRoom;

    /**消息类型*/
    private String msgType;

    /**消息内容*/
    private String msgContent;

    /**空构造方法*/
    public MessageDto() {
    }

    /**
     * 构造方法
     * @param sourceUserName 发起client
     * @param targetRoom 目标room
     * @param msgType 消息类型
     * @param msgContent 消息内容
     */
    public MessageDto(String sourceUserName, String targetRoom, String msgType, String msgContent) {
        this.sourceUserName = sourceUserName;
        this.targetRoom = targetRoom;
        this.msgType = msgType;
        this.msgContent = msgContent;
    }

    /**
     * json解析构造方法
     * @param messageJson json字符串
     * @throws Exception 异常
     */
    public MessageDto(String messageJson) throws Exception{
        MessageDto messageDto = new ObjectMapper().readValue(messageJson, MessageDto.class);
        this.sourceUserName = messageDto.getSourceUserName();
        this.targetRoom = messageDto.getTargetRoom();
        this.msgType = messageDto.getMsgType();
        this.msgContent = messageDto.getMsgContent();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

5. socketHandler

socketHandler内包含socket的connect、disconnect、event及message等信息。

• clientMap<String,List>是ConcurrentHashMap类，key为roomId，value为sessionId的list集合。用于存room（房间）空间内的所有client对应的sessionID。
• OnConnect
  
  OnConnect是客户端连接webSocket的触发事件，可以做校验、clientMap的put等。
• OnDisconnect
  
  客户端断开连接是调用，刷新客户端信息。
• OnEvent
  
  用于监听客户端emit的事件，并发送事件sendEvent。

6. 模拟客户端，代码如下：

public static void main(String[] args) {
        //服务端socket.io连接通信地址
        String url = "http://127.0.0.1:8083";
        try {
            IO.Options options = new IO.Options();
            options.transports = new String[]{"websocket"};
            options.reconnectionAttempts = 2;
            //失败重连的时间间隔
            options.reconnectionDelay = 1000;
            //连接超时时间(ms)
            options.timeout = 500;
            //userId: 唯一标识 传给服务端存储
            final Socket socket = IO.socket(url + "?userName=dong&appKey=abc-56&roomId=abc-56-1", options);
            socket.on(Socket.EVENT_CONNECT, args1 -> System.out.println(JSON.toJSONString(args1)));
            socket.on("ewsSocketMsg", objects -> System.out.println("服务端dong:" + objects[0].toString()));
            socket.on("ewsAllClientsMsg", objects -> System.out.println("服务端dong:" + objects[0].toString()));
            socket.connect();

            while (true) {
                Thread.sleep(10000);
                //自定义事件`push_data_event` -> 向服务端发送消息
                MessageDto messageDto = new MessageDto();
                messageDto.setSourceUserName("lipenghui");
                messageDto.setTargetRoom("abc-56-1");
                messageDto.setMsgType("2");
                messageDto.setMsgContent("ceshi guangbo");
                socket.emit("ewsSocketMsg", JSON.toJSONString(messageDto));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

整理完毕！