【实现方案】springboot 使用 websocket 与客户端实时通信（with 多人聊天室 + 心跳检测）_springboot websocket 心跳检测

前言

在之前一个项目中，有一个实时通信的需求需要实现，由此我初步接触了 websocket 协议，并做了简单的业务实现。作为一个初步的了解吧，给大家一个参考。

1. 为什么用 websocket？

换句话说，websocket 解决了什么问题？答案是，解决了两个主要问题：

• 只能客户端发送请求
• 一段时间内的频繁信息发送

假设现在需要设计一个实时预警系统的通知模块，那么作为工程师我们应该怎么设计通知的这个功能呢？如果我们现在只有 http 协议，那么我们只能让客户端不断地轮询服务器，轮询的时间间隔越小越能接近实时的效果。可是，轮询的效率低，又浪费资源。针对这样的场景，websocket 应运而生。

2. websocket

WebSocket 协议在2008年诞生，2011年成为国际标准。所有浏览器都已经支持了。
它的最大特点就是，服务器可以主动向客户端推送信息，客户端也可以主动向服务器发送信息，是真正的双向平等对话，属于服务器推送技术的一种。

其他特点包括：
（1）建立在 TCP 协议之上，服务器端的实现比较容易，是一个可靠的传输协议。
（2）与 HTTP 协议有着良好的兼容性。默认端口也是80和443，并且握手阶段采用 HTTP 协议，因此握手时不容易屏蔽，能通过各种 HTTP 代理服务器。
（3）数据格式比较轻量，性能开销小，通信高效。
（4）可以发送文本，也可以发送二进制数据。
（5）没有同源限制，客户端可以与任意服务器通信。
（6）协议标识符是ws（如果加密，则为wss），服务器网址就是 URL。

以上内容，摘自 http://www.ruanyifeng.com/blog/2017/05/websocket.html

3. websocket 在 springboot 中的实现

直接上代码，省略了 package 与 import，实现了一个多人多聊天室的 demo。

其中 onOpen 方法的并发处理与 webSocketSet 数据结构需要好好体会。

/**
 * websocket，实时信息回传。
 *
 * @author amber
 * @date 2020-08-05
 */

@Component
@ServerEndpoint("/websocket/{uniCode}") //通过 unicode 识别加入的用户（连接），放在一个集合中。
//每一个连接，就是一个实例。
public class RealTimeWebSocketServer{


    /**
     * 与某个客户端的连接对话，需要通过它来给客户端发送消息。
     * 每个用户（连接）私有。
     */
    private Session session;

    /**
     * static
     * 储存链接 <uniCode, set<instance>>
     * ConcurrentHashMap 线程安全，读取非阻塞
     * CopyOnWriteArraySet 线程安全
     */
    private static final ConcurrentHashMap<String, CopyOnWriteArraySet<RealTimeWebSocketServer>> webSocketSet = new ConcurrentHashMap<>();

	//每一个新连接进来执行的方法
    @OnOpen
    public void onOpen(Session session, @PathParam(value = "uniCode") String uniCode) {
        this.session = session;
        //先查找是否有uniCode
        CopyOnWriteArraySet<RealTimeWebSocketServer> users = webSocketSet.get(uniCode);
        if (users == null) {
            //处理读并发
            synchronized (webSocketSet) {
                if (!webSocketSet.contains(uniCode)) {
                    users = new CopyOnWriteArraySet<>();
                    webSocketSet.put(uniCode, users);
                    generateRealTimeInfo(uniCode);
                }
            }
        }
        users.add(this);
        logger.info("连接成功，当前房间数为：" + webSocketSet.size()
                + "，连接ID：" + uniCode
                + "，房间号：" + uniCode
                + "，当前房间人数：" + webSocketSet.get(uniCode).size());
    }

	//关闭连接执行的方法
    @OnClose
    public void onClose(Session session) {
        // 避免多人同时在线直接关闭通道。
        Object[] objects = webSocketSet.get(this.uniCode).toArray();
        for (int i = 0; i < objects.length; i++) {
            if(((RealTimeWebSocketServer) objects[i]).session.equals(session)){
                //删除房间中当前用户
                webSocketSet.get(this.uniCode).remove((RealTimeWebSocketServer) objects[i]);
            }
        }
        if(webSocketSet.get(uniCode).size() <= 0){
            //删除房间
            webSocketSet.remove(uniCode);
            logger.info("ID：" + uniCode+ "退出成功 ");
        }else{
            logger.info("ID：" + uniCode+ " 1名用户退出，剩余" + webSocketSet.get(uniCode).size() + "名用户在线");
        }
    }

	//实例收到消息执行此方法
    @OnMessage
    public void onMessage(String message) {
        logger.info("收到消息：" + message);
        //刷新实时巡检进度
        CopyOnWriteArraySet<RealTimeWebSocketServer> users = webSocketSet.get(uniCode);
        if (users != null) {
            for (RealTimeWebSocketServer user : users) {
                    user.session.getAsyncRemote().sendText(message);
                    logger.info("发送消息：" + message);
            }
        }
    }

	//未知错误执行此方法
    @OnError
    public void onError(Session session, Throwable error) {
        logger.info("发生错误" + new Date());
        logger.error(error.getMessage(), error);
    }

}
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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

4. 心跳检测

心跳检测是啥？

一般在长连接中，有可能很长一段时间都没有数据往来，所以一般周期性的，前端会发送一个心跳包到后端，保证连接的有效性。

为啥不是后端发心跳检测，而是前端发？

• 后端可能保有大量的连接，如果作为定时任务对无数据连接发送心跳包，开销大。
• 从业务角度来说，每一个前端作为大部分信息的发送和接收方，控制连接更为合理。

具体实现

应用层，这是一个前后端相互配合的功能。

前端代码

//心跳检测
var heartCheck = {
      timeout: 60000, //每隔60秒发送心跳
      severTimeout: 5000,  //服务端超时时间
      timeoutObj: null,
      serverTimeoutObj: null,
      start: function(){
        var _this = this;
        this.timeoutObj && clearTimeout(this.timeoutObj);
        this.serverTimeoutObj && clearTimeout(this.serverTimeoutObj);
        this.timeoutObj = setTimeout(function(){
              //这里发送一个心跳，后端收到后，返回一个心跳消息，
              //onmessage拿到返回的心跳就说明连接正常
              ws.send("心跳包"); // 心跳包
              //计算答复的超时时间
              _this.serverTimeoutObj = setTimeout(function() {
                  ws.close();
              }, _this.severTimeout);
        }, this.timeout)
      }
}

====================================================================
  ws.onopen = function () {
    //心跳检测重置
    heartCheck.start();
  };
  
  ws.onmessage = function (event) {
    //拿到任何消息都说明当前连接是正常的
    console.log('接收到消息');
    heartCheck.start();
  }
  
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

参考资料

https://www.cnblogs.com/FatKee/articles/10250854.html