【分布式websocket】IM系统中为什么要做心跳！Netty中如何做心跳，客户端如何做断线重连！_netty websocket心跳

前言

B站视频版本

在IM（即时通讯）聊天系统中，"心跳"是一个非常关键且生动的概念。想象一下，就像我们人类的心脏每时每刻都在跳动以证明我们的生命活力一样，在网络世界里，"心跳机制"也扮演着维持“活跃状态”和“连接健康”的角色。

当你使用qq微信时候总不能网络一不好就自动掉线吧。所以后台有一套心跳机制。 即使网络波动等因素也可以继续聊天使用软件。

"心跳机制"就是为了解决这个问题而存在的。简单来说，它就像是你设备向服务器发送的一个个“我在呢”、“我还在线”的信号。每隔一定时间（比如几秒钟），你的设备就会自动向服务器发送一个小小的数据包，这个数据包我们就形象地称之为“心跳包”。

服务器接收到这个“心跳包”后，就知道你的设备依然在线，并且网络连接是正常的。如果服务器在一段时间内没有接收到某个用户的心跳包，那么就可以判断该用户的连接可能出现问题，从而采取相应的措施，如断开并重新建立连接，或者标记用户为离线状态。

目前已经写的文章有。并且有对应视频版本。
git项目地址 【IM即时通信系统（企聊聊）】点击可跳转
sprinboot单体项目升级成springcloud项目 【第一期】
分布式权限 shiro + jwt + redis【第三期】
分布式websocket即时通信(IM)系统构建指南【第七期】
分布式websocket即时通信(IM)系统保证消息可靠性【第八期】
分布式websocket IM聊天系统相关问题问答【第九期】
什么？websocket也有权限！这个应该怎么做？【第十期】
分布式ID是什么，以美团Leaf为例改造融入自己项目【第十一期】
IM聊天系统为什么需要做消息幂等？如何使用Redis以及Lua脚本做消息幂等【第12期】
微信发送一条消息经历哪些过程。企业微信以及钉钉的IM架构对比【第13期】
微信群为什么上限是500人，IM设计系统中的群聊的设计难点【第14期】
【分布式websocket】RocketMQ发送消息保证消息最终一致性需要做哪些处理？【第15期】

【分布式websocket】群聊中的各种难点以及解决推拉结合【第16期】

【分布式webscoket】未读消息如何设计？解决缓存与数据库数据一致性！推送未读消息流程【第17期】

心跳过程

心跳之际对于节省服务端资源，保证客户端存活等有着重要作用。
**服务端主动探测：**每间隔一定时间后，向所有客户端发送一个检测信号，过程如下：

假设目前有三个节点，A为服务端，B、C都为客户端。
A：你们还活着吗？
B：我还活着！
C：…（假设挂掉了，无响应）

A收到了B的响应，但C却未给出响应，很有可能挂了，A中断与C的连接。

**客户端主动告知：**每间隔一定时间后，客户端向服务端发送一个心跳包，过程如下：
依旧是上述那三个节点。
B：我还活着，不要开除我！
C：…（假设挂掉了，不发送心跳包）
A：收到B的心跳包，但未收到C的心跳包，将C的网络连接断开。

一般来说，一套健全的心跳机制，都会结合上述两种方案一起实现，也就是客户端定时向服务端发送心跳包，当服务端未收到某个客户端心跳包的情况下，再主动向客户端发起探测包，这一步主要是做二次确认，防止由于网络拥塞或其他问题，导致原本客户端发出的心跳包丢失。

netty心跳

心跳
Netty 的超时类型 IdleState 主要分为以下3类：

ALL_IDLE : 一段时间内没有数据接收或者发送。
READER_IDLE ： 一段时间内没有数据接收。
WRITER_IDLE ： 一段时间内没有数据发送。

服务端代码处理

  //单位是秒
    private int READER_IDLE_TIME = 120;
    private int  WRITER_IDLE_TIME = 60;

    private int ALL_IDLE_TIME = 180;
    @Override
    protected void initChannel(SocketChannel e) throws Exception {
        e.pipeline().addLast("http-codec", new HttpServerCodec()) //http编解码
        /**

         *HttpObjectAggregator 因为http在传输过程中是分段的，HttpObjectAggregator可以将多个段聚合起来
         * 这就是为什么当浏览器发送大量数据时，会发出多次http请求
         */
                    .addLast("aggregator",new HttpObjectAggregator(65536)) //httpContent消息聚合
                    .addLast("http-chunked",new ChunkedWriteHandler())  // HttpContent 压缩
                /**

                 *WebSocketServerProtocolHandler 对应websocket，它的数据是以 帧(frame)形式 传递
                 * 可以看到 WebSocketFrame 下有六个子类
                 * 浏览器请求时，ws://localhost:7000/XXX 表示请求的资源
                 * 核心功能是 将http协议升级为ws协议，保持长连接
                 */
                    .addLast("nettyWebSocketParamHandler",nettyWebSocketParamHandler)
                    .addLast("protocolHandler",new WebSocketServerProtocolHandler("/websocket"))
                    .addLast(new IdleStateHandler(READER_IDLE_TIME,
                            WRITER_IDLE_TIME,
                            ALL_IDLE_TIME,
                            TimeUnit.SECONDS))
                    .addLast("base_handler",myWebSocketHandler)
                    .addLast("register_handler",registerHandler)
                    .addLast("single_message",singleMessageHandler)
                     .addLast("ack_single_message",ackSingleMessageHandler)
                    .addLast("creat_group",creatGroupHandler)
                    .addLast("group_message",groupMessageHandler)
//                    .addLast(HeartBeatRequestHandler.INSTANCE)
                    .addLast(ExceptionHandler.INSTANCE);
    }
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

引入心跳。设定好时间。然后再userEventTriggered里面做心跳相关的逻辑。当客户端不跳的时候咱就给他停了，干脆一直别跳了。一方面关掉netty里面的channel通道。另一方面要清除redis里面的客户端信息。

 if(evt instanceof IdleStateEvent) {
            //将  evt 向下转型 IdleStateEvent
            IdleStateEvent event = (IdleStateEvent) evt;
            String eventType = null;
            switch (event.state()) {
                case READER_IDLE:
                    eventType = "读空闲";
                    channelService.remove(openid2);
                    log.info(ctx.channel().remoteAddress() + "--超时时间--" + eventType);
                    ctx.channel().close();

                    //关闭通道，并且清除所有的redis信息
                    break;
                case WRITER_IDLE:
                    eventType = "写空闲";
                    break;
                case ALL_IDLE:
                    eventType = "读写空闲";
                    break;
            }
            //这里已经可以知道浏览器所处的空闲是何种空闲，可以执行对应的处理逻辑了

        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

 /**
     * 移除客户端
     * @param
     */
    public void remove(String channelId){

        String instanceid = nettyutil.getInstance();

//        if(!StringUtils.isNotEmpty(instanceid)){return;}
//        get(instanceid);
        Channel channel = SessionUtils.getChannel(channelId);
//        if(channel==null){return;}
        try {
            String dateTime = channel.attr(AttrConstants.activeTime).get();
            //断开当前连接
//            get(instanceid).close();
//            channel.closeFuture().addListener()
            //删除redis中维护的客户端信息
            redisTemplate.delete(RedisPrefix.PREFIX_CLIENT+channelId);
            //删除redis中客户端与host的关联关系
            redisTemplate.opsForSet().remove(RedisPrefix.PREFIX_SERVERCLIENTS+instanceid,channelId);
            log.info("移除了客户端[{}],上一次的活跃时间为[{}]",
                    channelId,
                    StringUtils.isNotEmpty(dateTime)? DateUtils.dateToDateTime(new Date(Long.parseLong(dateTime))):"");
        }catch (Exception e){
            log.error("移除客户端失败["+instanceid+"]",e);
        }
    }
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

主要是删除redis

客户端心跳逻辑

export default class SocketService {
  static instance = null;
  static get Instance() {
    if (!this.instance) {
      this.instance = new SocketService();
    }
    return this.instance;
  }
  // 和服务端连接的socket对象
  ws = null;
  // 存储回调函数
  callBackMapping = {};
  // 标识是否连接成功
  connected = false;
  // 记录重试的次数
  sendRetryCount = 0;
  // 重新连接尝试的次数
  connectRetryCount = 0;
  openidself = null;
  //  定义连接服务器的方法
  // 定义发送ping的方法
  sendPing() {
    if (this.ws && this.connected) {
      const pingMessage = { type: "20" }; // 或者其他格式，取决于后端需要的ping消息格式
      this.ws.send(JSON.stringify(pingMessage)); // 发送ping数据
    }
  }

  // 在建立连接成功后启动心跳定时器
  startHeartbeatInterval() {
    if (this.ws) {
      this.pingIntervalId = setInterval(() => this.sendPing(), 30000); // 设置间隔为30秒发送一次ping，你可以根据实际需求调整这个时间
    }
  }
  connect(token, openid) {
    // 连接服务器
    if (!window.WebSocket) {
      return console.log("您的浏览器不支持WebSocket");
    }
    // let token = $.cookie('123');
    // let token = '4E6EF539AAF119D82AC4C2BC84FBA21F';
    let url = "ws://127.0.1:88/websocket?token=" + token + "&openid=" + openid;
    this.openidself = openid;
    this.ws = new WebSocket(url);
    // 连接成功的事件
    this.ws.onopen = () => {
      console.log("【IM日志】连接服务端成功了");
      this.connected = true;
      this.startHeartbeatInterval(); // 连接成功后启动心跳
      // 重置重新连接的次数
      //连接成功并且返回正确结果 才能重置连接次数
      this.connectRetryCount = 0;
      console.log("onopen 重连次数", this.connectRetryCount);
    };
    // 1.连接服务端失败
    // 2.当连接成功之后, 服务器关闭的情况  自动重连五次.
    this.ws.onclose = () => {
      console.log("【IM日志】连接服务端失败");
      this.connected = false;
      this.connectRetryCount++;
      console.log("onclose", this.connectRetryCount);
      if (this.connectRetryCount == 5) {
        return;
      }
      //断开连接的时候清除定时器. 
      if (this.pingIntervalId) {
        clearInterval(this.pingIntervalId);
      }
      setTimeout(() => {
        this.connect(localStorage.getItem("token"), this.openidself);
      }, 1000 * this.connectRetryCount);
    };
    // 得到服务端发送过来的数据
    // this.ws.onmessage = msg => {
    //   // console.log(msg.data, '从服务端获取到了数据');
    // };
  }
  // 回调函数的注册
  registerCallBack(socketType, callBack) {
    this.callBackMapping[socketType] = callBack;
  }
  // 取消某一个回调函数
  unRegisterCallBack(socketType) {
    this.callBackMapping[socketType] = null;
  }
  // 发送数据的方法
  send(data) {
    // 判断此时此刻有没有连接成功
    if (this.connected) {
      this.sendRetryCount = 0;
      try {
        this.ws.send(JSON.stringify(data));
      } catch (e) {
        this.ws.send(data);
      }
    } else {
      this.sendRetryCount++;
      setTimeout(() => {
        this.send(data);
      }, this.sendRetryCount * 500);
    }
  }
}

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
93
94
95
96
97
98
99
100
101
102
103
104

sendPing和 startHeartbeatInterval 是用来发送心跳的客户端逻辑。

  case 20:
                ByteBuf buf = createPongByteBuf(ctx);
                 TextWebSocketFrame tws = new TextWebSocketFrame(buf);
                ctx.writeAndFlush(tws);
                log.info("收到了ping");
                break;


    public ByteBuf createPongByteBuf(ChannelHandlerContext ctx) {
        ByteBuf byteBuf = ctx.alloc().buffer();

        JSONObject data = new JSONObject();
        JSONObject params = new JSONObject();
        params.put("type", "pong");
        params.put("date", new Date().toString());
        data.put("params", params);
        byte []bytes = data.toJSONString().getBytes(Charset.forName("utf-8"));
        byteBuf.writeBytes(bytes);
        return byteBuf;
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

客户端相应并且回复pong；

断线重连

// 1.连接服务端失败
    // 2.当连接成功之后, 服务器关闭的情况  自动重连五次.
    this.ws.onclose = () => {
      console.log("【IM日志】连接服务端失败");
      this.connected = false;
      this.connectRetryCount++;
      console.log("onclose", this.connectRetryCount);
      if (this.connectRetryCount == 5) {
        return;
      }
      //断开连接的时候清除定时器. 
      if (this.pingIntervalId) {
        clearInterval(this.pingIntervalId);
      }
      setTimeout(() => {
        this.connect(localStorage.getItem("token"), this.openidself);
      }, 1000 * this.connectRetryCount);
    };
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

就是当这个服务器断开连接的时候写了一个定时器去调用连接的函数来自动重连。

不过上述的心跳机制仅实现了最基础的版本，还未彻底将其完善，但我这里就不继续往下实现了，毕竟主干已经搭建好了，剩下的只是一些细枝末节，我这里提几点完善思路：

①在检测到某个客户端未发送心跳包的情况下，服务端应当主动再发起一个探测包，二次确认客户端是否真的挂了，这样做的好处在于：能够有效避免网络抖动造成的“客户端假死”现象。
②客户端、服务端之间交互的数据包，应当采用统一的格式进行封装，也就是都遵守同一规范包装数据，例如{msgType:“Heartbeat”, msgContent:“…”, …}。
③在客户端被关闭的情况下，但凡不是因为物理因素，如机房断电、网线被拔、机器宕机等情况造成的客户端下线，客户端都必须具备断线重连功能。

将上述三条完善后，才能够被称为是一套相对健全的心跳检测机制，所以大家感兴趣的情况下，可基于前面给出的源码接着实现~