netty heartbeat机制

逃离我推掉我的手 2024-04-17 21:18 103阅读 0赞

# 前提： #

本文会直接使用netty的LineBasedFrameDecoder和StringDecoder作为例子，介绍heatbeat机制，因此读者需要具备一定netty编程基础。

# 问题由来 #

所谓心跳, 就在在 TCP长连接中, 客户端和服务器之间定期发送的一种特殊数据包, 通知对方自己还在线, 以确保 TCP 连接的有效性。

因为网络的不可靠性, 有可能在 TCP 保持长连接的过程中, 由于某些突发情况, 例如网线被拔出, 突然掉电等, 会造成服务器和客户端的连接中断. 在这些突发情况下, 如果恰好服务器和客户端之间没有交互的话, 那么它们是不能在短时间内发现对方已经掉线的. 为了解决这个问题, 我们就需要引入心跳机制.

心跳机制的工作原理是: 在服务器和客户端之间一定时间内没有数据交互时, 即处于 idle 状态时, 客户端或服务器会发送一个特殊的数据包给对方, 当接收方收到这个数据报文后, 也立即发送一个特殊的数据报文, 回应发送方, 此即一个 PING-PONG 交互. 自然地, 当某一端收到心跳消息后, 就知道了对方仍然在线, 这就确保 TCP 连接的有效性.  
如何实现心跳

两种方式实现心跳机制:  
使用 TCP 协议层面的 keepalive 机制.  
在应用层上实现自定义的心跳机制.

虽然在 TCP 协议层面上, 提供了 keepalive 保活机制, 但是使用它有几个缺点:  
它不是 TCP 的标准协议, 并且是默认关闭的.  
TCP keepalive 机制依赖于操作系统的实现, 默认的 keepalive 心跳时间是 两个小时, 并且对 keepalive 的修改需要系统调用(或者修改系统配置), 灵活性不够.

TCP keepalive 与 TCP 协议绑定, 因此如果需要更换为 UDP 协议时, keepalive 机制就失效了.

基于以上缺点, 一般实践中, 人们大多数都是选择在应用层上实现自定义的心跳.

# 解决办法 #

Netty已经为我们提供了如何实现心跳功能的办法， 这就是IdleStateHandler。 在 Netty 中, 实现心跳机制的关键是 IdleStateHandler, 它可以对一个 Channel 的 读/写设置定时器, 当 Channel 在一定事件间隔内没有数据交互时(即处于 idle 状态), 就会触发指定的事件.

# 最终实现 #

我的例子比较简单，就是服务器端如果发现某个客户端连续3次（idle时间第5秒）都没有发送数据，就断开连接。 也就是如果某个连接已经超过15秒都不想服务器上传消息，服务器就认为该客户端**异常**

通常空闲时间以及异常是根据业务定义的。 我们的业务很简单，服务器给让客户端执行job， 客户端边执行边发送结果。如果客户端没有job执行，也有需要不断向服务器发送ping，表示自己还在active状态，这样服务器端就修改该客户端的状态为ready，一旦条件符合就选择该客户端执行任务。（现在的示例中，客户端没有检测服务器端是否正常，只是保证自己（client端），如果没有job日志需要发送时，定期（也就是写空闲时）向服务器端发送心跳）。 示例比较简单，读者需要根据自己的业务进行判断和修改，但基本用法是一致。 本文示例直接使用行分隔符作为消息解码器，实际当中我们使用的protobuf相关的decoder和encoder。为了示例简单易懂，就使用最简单的LineBasedFrameDecoder

完整代码在[这里][Link 1]， 欢迎fork， 加星。 谢谢！

import com.yq.uitl.SocketUtils;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.handler.timeout.IdleState;
    import io.netty.handler.timeout.IdleStateEvent;
    import lombok.NoArgsConstructor;
    import lombok.extern.slf4j.Slf4j;
    
    @Slf4j
    @NoArgsConstructor
    public class ServerSideHandler extends SimpleChannelInboundHandler<String> { 
        private int idleCounter = 0;
    
        @Override
        public void channelActive(final ChannelHandlerContext ctx) { 
            log.info("---Connection Created from {}", ctx.channel().remoteAddress());
            SocketUtils.sendHello(ctx, "server", false);
        }
    
        @Override
        public void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { 
            // Send the received message to all channels but the current one.
            log.info("ip:{}--- msg:{}", ctx.channel().remoteAddress(), msg);
            idleCounter = 0;;
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
            log.warn("Unexpected exception from downstream.", cause);
            ctx.close();
        }
    
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt)
                throws Exception { 
            if (evt instanceof IdleStateEvent) { 
                IdleStateEvent event = (IdleStateEvent) evt;
                if (event.state().equals(IdleState.READER_IDLE)) { 
                    log.warn("第" + idleCounter +"次没收到客户端信息了。ip={}", ctx.channel().remoteAddress());
                    if (idleCounter > 3) { 
                        // 超时关闭channel
                        log.warn("已经连续三次没收到客户端信息了， 关闭不活跃的连接={}", ctx);
                        ctx.close();
                    } else { 
                        idleCounter++;
                    }
                } else if (event.state().equals(IdleState.WRITER_IDLE)) { 
                    log.info("写空闲");
                } else if (event.state().equals(IdleState.ALL_IDLE)) { 
                    log.info("ALL_IDLE");
                    // 发送心跳
                    ctx.channel().write("ping\n");
                }
            }
            super.userEventTriggered(ctx, evt);
        }
    }

客户端检测写空闲，发现自己已经8秒没有写消息给服务器，就发送一个ping消息到服务器。

package com.yq.client;
    
    
    import com.yq.uitl.SocketUtils;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.handler.timeout.IdleState;
    import io.netty.handler.timeout.IdleStateEvent;
    import lombok.NoArgsConstructor;
    import lombok.extern.slf4j.Slf4j;
    
    @Slf4j
    @NoArgsConstructor
    public class ClientSideHandler extends SimpleChannelInboundHandler<String> { 
        private int idleCounter = 0;
    
        @Override
        public void channelActive(final ChannelHandlerContext ctx) { 
            System.out.println("connected");
            log.info("---Connection Created from {}", ctx.channel().remoteAddress());
            //SocketUtils.sendHello(ctx,"Client", false);
    
            String str20 = "012345678901234567890123456789";
            SocketUtils.sendLineBaseText(ctx, str20);
        }
    
        @Override
        public void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception { 
            // Send the received message to all channels but the current one.
            log.info("ip={}--- msg={}", ctx.channel().remoteAddress(), msg);
    
            idleCounter = 0;
            String str20 = "01234567890123456789";
            SocketUtils.sendLineBaseText(ctx, str20);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
            log.warn("Unexpected exception from downstream.", cause);
            ctx.close();
        }
    
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt)
                throws Exception { 
            if (evt instanceof IdleStateEvent) { 
                IdleStateEvent event = (IdleStateEvent) evt;
                if (event.state().equals(IdleState.READER_IDLE)) { 
                    log.warn("5秒没收到服务器端信息了.");
                } else if (event.state().equals(IdleState.WRITER_IDLE)) { 
                    log.warn("第" + idleCounter +"次没向服务器端发送信息了。ip={}", ctx.channel().remoteAddress());
                    if (idleCounter > 1) { 
                        log.warn("向服务器端发送一次心跳");
                        // 发送心跳
                        SocketUtils.sendLineBaseText(ctx, "ping");
                        idleCounter = 0;
                    } else { 
                        idleCounter++;
                    }
                } else if (event.state().equals(IdleState.ALL_IDLE)) { 
                    log.info("ALL_IDLE");
                    // 发送心跳
                     SocketUtils.sendLineBaseText(ctx, "ping");
                }
            }
            super.userEventTriggered(ctx, evt);
        }
    
    }

执行结果  
服务器端日志

[INFO ] 2019-09-01 12:51:30,732 [ nioEventLoopGroup-3-1:4299 ] method:com.yq.server.ServerSideHandler.channelRead0(ServerSideHandler.java:27)
    ip:/192.168.1.104:60581--- msg:012345678901234567890123456789
    [INFO ] 2019-09-01 12:51:30,735 [ nioEventLoopGroup-3-1:4302 ] method:com.yq.server.ServerSideHandler.channelRead0(ServerSideHandler.java:27)
    ip:/192.168.1.104:60581--- msg:01234567890123456789
    [WARN ] 2019-09-01 12:51:35,738 [ nioEventLoopGroup-3-1:9305 ] method:com.yq.server.ServerSideHandler.userEventTriggered(ServerSideHandler.java:43)
    第0次没收到客户端信息了。ip=/192.168.1.104:60581
    [WARN ] 2019-09-01 12:51:40,737 [ nioEventLoopGroup-3-1:14304 ] method:com.yq.server.ServerSideHandler.userEventTriggered(ServerSideHandler.java:43)
    第1次没收到客户端信息了。ip=/192.168.1.104:60581
    [WARN ] 2019-09-01 12:51:45,740 [ nioEventLoopGroup-3-1:19307 ] method:com.yq.server.ServerSideHandler.userEventTriggered(ServerSideHandler.java:43)
    第2次没收到客户端信息了。ip=/192.168.1.104:60581
    [INFO ] 2019-09-01 12:51:45,741 [ nioEventLoopGroup-3-1:19308 ] method:com.yq.server.ServerSideHandler.channelRead0(ServerSideHandler.java:27)
    ip:/192.168.1.104:60581--- msg:ping
    [WARN ] 2019-09-01 12:51:50,742 [ nioEventLoopGroup-3-1:24309 ] method:com.yq.server.ServerSideHandler.userEventTriggered(ServerSideHandler.java:43)
    第0次没收到客户端信息了。ip=/192.168.1.104:60581
    [WARN ] 2019-09-01 12:51:55,742 [ nioEventLoopGroup-3-1:29309 ] method:com.yq.server.ServerSideHandler.userEventTriggered(ServerSideHandler.java:43)
    第1次没收到客户端信息了。ip=/192.168.1.104:60581
    [WARN ] 2019-09-01 12:52:00,743 [ nioEventLoopGroup-3-1:34310 ] method:com.yq.server.ServerSideHandler.userEventTriggered(ServerSideHandler.java:43)
    第2次没收到客户端信息了。ip=/192.168.1.104:60581
    [INFO ] 2019-09-01 12:52:00,744 [ nioEventLoopGroup-3-1:34311 ] method:com.yq.server.ServerSideHandler.channelRead0(ServerSideHandler.java:27)
    ip:/192.168.1.104:60581--- msg:ping
    [WARN ] 2019-09-01 12:52:05,745 [ nioEventLoopGroup-3-1:39312 ] method:com.yq.server.ServerSideHandler.userEventTriggered(ServerSideHandler.java:43)
    第0次没收到客户端信息了。ip=/192.168.1.104:60581

客户端日志

[INFO ] 2019-09-01 12:51:30,734 [ nioEventLoopGroup-2-1:1335 ] method:com.yq.client.ClientSideHandler.channelRead0(ClientSideHandler.java:31)
    ip=/192.168.1.104:5566--- msg=HELLO from HeatBeatDemo server
    [INFO ] 2019-09-01 12:51:30,734 [ nioEventLoopGroup-2-1:1335 ] method:io.netty.util.internal.logging.Slf4JLogger.info(Slf4JLogger.java:101)
    [id: 0xe8f2e2a1, L:/192.168.1.104:60581 - R:/192.168.1.104:5566] WRITE: 22B
             +-------------------------------------------------+
             |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    +--------+-------------------------------------------------+----------------+
    |00000000| 30 31 32 33 34 35 36 37 38 39 30 31 32 33 34 35 |0123456789012345|
    |00000010| 36 37 38 39 0d 0a                               |6789..          |
    +--------+-------------------------------------------------+----------------+
    [INFO ] 2019-09-01 12:51:30,735 [ nioEventLoopGroup-2-1:1336 ] method:io.netty.util.internal.logging.Slf4JLogger.info(Slf4JLogger.java:101)
    [id: 0xe8f2e2a1, L:/192.168.1.104:60581 - R:/192.168.1.104:5566] FLUSH
    [INFO ] 2019-09-01 12:51:30,735 [ nioEventLoopGroup-2-1:1336 ] method:io.netty.util.internal.logging.Slf4JLogger.info(Slf4JLogger.java:101)
    [id: 0xe8f2e2a1, L:/192.168.1.104:60581 - R:/192.168.1.104:5566] READ COMPLETE
    [WARN ] 2019-09-01 12:51:35,737 [ nioEventLoopGroup-2-1:6338 ] method:com.yq.client.ClientSideHandler.userEventTriggered(ClientSideHandler.java:52)
    第0次没向服务器端发送信息了。ip=/192.168.1.104:5566
    [WARN ] 2019-09-01 12:51:40,736 [ nioEventLoopGroup-2-1:11337 ] method:com.yq.client.ClientSideHandler.userEventTriggered(ClientSideHandler.java:52)
    第1次没向服务器端发送信息了。ip=/192.168.1.104:5566
    [WARN ] 2019-09-01 12:51:45,736 [ nioEventLoopGroup-2-1:16337 ] method:com.yq.client.ClientSideHandler.userEventTriggered(ClientSideHandler.java:52)
    第2次没向服务器端发送信息了。ip=/192.168.1.104:5566
    [WARN ] 2019-09-01 12:51:45,737 [ nioEventLoopGroup-2-1:16338 ] method:com.yq.client.ClientSideHandler.userEventTriggered(ClientSideHandler.java:54)
    向服务器端发送一次心跳
    [INFO ] 2019-09-01 12:51:45,737 [ nioEventLoopGroup-2-1:16338 ] method:io.netty.util.internal.logging.Slf4JLogger.info(Slf4JLogger.java:101)
    [id: 0xe8f2e2a1, L:/192.168.1.104:60581 - R:/192.168.1.104:5566] WRITE: 6B
             +-------------------------------------------------+
             |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    +--------+-------------------------------------------------+----------------+
    |00000000| 70 69 6e 67 0d 0a                               |ping..          |
    +--------+-------------------------------------------------+----------------+
    [INFO ] 2019-09-01 12:51:45,738 [ nioEventLoopGroup-2-1:16339 ] method:io.netty.util.internal.logging.Slf4JLogger.info(Slf4JLogger.java:101)
    [id: 0xe8f2e2a1, L:/192.168.1.104:60581 - R:/192.168.1.104:5566] FLUSH
    [WARN ] 2019-09-01 12:51:50,739 [ nioEventLoopGroup-2-1:21340 ] method:com.yq.client.ClientSideHandler.userEventTriggered(ClientSideHandler.java:52)
    第0次没向服务器端发送信息了。ip=/192.168.1.104:5566
    [WARN ] 2019-09-01 12:51:55,739 [ nioEventLoopGroup-2-1:26340 ] method:com.yq.client.ClientSideHandler.userEventTriggered(ClientSideHandler.java:52)
    第1次没向服务器端发送信息了。ip=/192.168.1.104:5566
    [WARN ] 2019-09-01 12:52:00,740 [ nioEventLoopGroup-2-1:31341 ] method:com.yq.client.ClientSideHandler.userEventTriggered(ClientSideHandler.java:52)
    第2次没向服务器端发送信息了。ip=/192.168.1.104:5566
    [WARN ] 2019-09-01 12:52:00,741 [ nioEventLoopGroup-2-1:31342 ] method:com.yq.client.ClientSideHandler.userEventTriggered(ClientSideHandler.java:54)
    向服务器端发送一次心跳
    [INFO ] 2019-09-01 12:52:00,742 [ nioEventLoopGroup-2-1:31343 ] method:io.netty.util.internal.logging.Slf4JLogger.info(Slf4JLogger.java:101)
    [id: 0xe8f2e2a1, L:/192.168.1.104:60581 - R:/192.168.1.104:5566] WRITE: 6B
             +-------------------------------------------------+
             |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    +--------+-------------------------------------------------+----------------+
    |00000000| 70 69 6e 67 0d 0a                               |ping..          |
    +--------+-------------------------------------------------+----------------+
    [INFO ] 2019-09-01 12:52:00,742 [ nioEventLoopGroup-2-1:31343 ] method:io.netty.util.internal.logging.Slf4JLogger.info(Slf4JLogger.java:101)
    [id: 0xe8f2e2a1, L:/192.168.1.104:60581 - R:/192.168.1.104:5566] FLUSH
    [WARN ] 2019-09-01 12:52:05,743 [ nioEventLoopGroup-2-1:36344 ] method:com.yq.client.ClientSideHandler.userEventTriggered(ClientSideHandler.java:52)
    第0次没向服务器端发送信息了。ip=/192.168.1.104:5566

如果我们使用telnet连接服务器，会发现因为我们在telent中有超过15秒不发消息，telent被自动中断  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1c3NsZQ_size_16_color_FFFFFF_t_70]

[Link 1]: https://github.com/yqbjtu/springboot/tree/master/nettyHeatBeat
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1c3NsZQ_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/ff7813a8cd65469495465581100d42d5.png