用Netty撸一个心跳机制和断线重连！

迈不过友情╰ 2023-10-04 09:24 11阅读 0赞

来源：www.jianshu.com/p/1a28e48edd92

### 心跳机制 ###

#### 何为心跳 ####

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

> 注：心跳包还有另一个作用，经常被忽略，即：**一个连接如果长时间不用，防火墙或者路由器就会断开该连接**。

### 如何实现 ###

#### 核心Handler —— IdleStateHandler ####

在 `Netty` 中, 实现心跳机制的关键是 `IdleStateHandler`, 那么这个 `Handler` 如何使用呢? 先看下它的构造器：

public IdleStateHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) {
        this((long)readerIdleTimeSeconds, (long)writerIdleTimeSeconds, (long)allIdleTimeSeconds, TimeUnit.SECONDS);
    }

这里解释下三个参数的含义：

*  readerIdleTimeSeconds: 读超时. 即当在指定的时间间隔内没有从 `Channel` 读取到数据时, 会触发一个 `READER_IDLE` 的 `IdleStateEvent` 事件.
 *  writerIdleTimeSeconds: 写超时. 即当在指定的时间间隔内没有数据写入到 `Channel` 时, 会触发一个 `WRITER_IDLE` 的 `IdleStateEvent` 事件.
 *  allIdleTimeSeconds: 读/写超时. 即当在指定的时间间隔内没有读或写操作时, 会触发一个 `ALL_IDLE` 的 `IdleStateEvent` 事件.

> 注：这三个参数默认的时间单位是**秒**。若需要指定其他时间单位，可以使用另一个构造方法：`IdleStateHandler(boolean observeOutput, long readerIdleTime, long writerIdleTime, long allIdleTime, TimeUnit unit)`

在看下面的实现之前，建议先了解一下`IdleStateHandler`的实现原理。

下面直接上代码，需要注意的地方，会在代码中通过注释进行说明。

#### 使用IdleStateHandler实现心跳 ####

下面将使用`IdleStateHandler`来实现心跳，`Client`端连接到`Server`端后，会循环执行一个任务：**随机等待几秒，然后`ping`一下`Server`端，即发送一个心跳包**。当等待的时间超过规定时间，将会发送失败，以为`Server`端在此之前已经主动断开连接了。代码如下：

##### Client端 #####

###### ClientIdleStateTrigger —— 心跳触发器 ######

类`ClientIdleStateTrigger`也是一个`Handler`，只是重写了`userEventTriggered`方法，用于捕获`IdleState.WRITER_IDLE`事件（未在指定时间内向服务器发送数据），然后向`Server`端发送一个心跳包。

/**
  * 
  *  用于捕获{@link IdleState#WRITER_IDLE}事件（未在指定时间内向服务器发送数据），然后向<code>Server</code>端发送一个心跳包。
  * 
  */
 public class ClientIdleStateTrigger extends ChannelInboundHandlerAdapter {
 
     public static final String HEART_BEAT = "heart beat!";
 
     @Override
     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
         if (evt instanceof IdleStateEvent) {
             IdleState state = ((IdleStateEvent) evt).state();
             if (state == IdleState.WRITER_IDLE) {
                 // write heartbeat to server
                 ctx.writeAndFlush(HEART_BEAT);
             }
         } else {
             super.userEventTriggered(ctx, evt);
         }
     }
 
 }

###### Pinger —— 心跳发射器 ######

/**
  * 客户端连接到服务器端后，会循环执行一个任务：随机等待几秒，然后ping一下Server端，即发送一个心跳包。
  */
 public class Pinger extends ChannelInboundHandlerAdapter {
 
     private Random random = new Random();
     private int baseRandom = 8;
 
     private Channel channel;
 
     @Override
     public void channelActive(ChannelHandlerContext ctx) throws Exception {
         super.channelActive(ctx);
         this.channel = ctx.channel();
 
         ping(ctx.channel());
     }
 
     private void ping(Channel channel) {
         int second = Math.max(1, random.nextInt(baseRandom));
         System.out.println("next heart beat will send after " + second + "s.");
         ScheduledFuture<?> future = channel.eventLoop().schedule(new Runnable() {
             @Override
             public void run() {
                 if (channel.isActive()) {
                     System.out.println("sending heart beat to the server...");
                     channel.writeAndFlush(ClientIdleStateTrigger.HEART_BEAT);
                 } else {
                     System.err.println("The connection had broken, cancel the task that will send a heart beat.");
                     channel.closeFuture();
                     throw new RuntimeException();
                 }
             }
         }, second, TimeUnit.SECONDS);
 
         future.addListener(new GenericFutureListener() {
             @Override
             public void operationComplete(Future future) throws Exception {
                 if (future.isSuccess()) {
                     ping(channel);
                 }
             }
         });
     }
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
         // 当Channel已经断开的情况下, 仍然发送数据, 会抛异常, 该方法会被调用.
         cause.printStackTrace();
         ctx.close();
     }
 }

###### ClientHandlersInitializer —— 客户端处理器集合的初始化类 ######

public class ClientHandlersInitializer extends ChannelInitializer<SocketChannel> {
 
     private ReconnectHandler reconnectHandler;
     private EchoHandler echoHandler;
 
     public ClientHandlersInitializer(TcpClient tcpClient) {
         Assert.notNull(tcpClient, "TcpClient can not be null.");
         this.reconnectHandler = new ReconnectHandler(tcpClient);
         this.echoHandler = new EchoHandler();
     }
 
     @Override
     protected void initChannel(SocketChannel ch) throws Exception {
         ChannelPipeline pipeline = ch.pipeline();
         pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
         pipeline.addLast(new LengthFieldPrepender(4));
         pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
         pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
         pipeline.addLast(new Pinger());
     }
 }

> 注：上面的`Handler`集合，除了`Pinger`，其他都是编解码器和解决粘包，可以忽略。

###### TcpClient —— TCP连接的客户端 ######

public class TcpClient {
 
     private String host;
     private int port;
     private Bootstrap bootstrap;
     /** 将<code>Channel</code>保存起来, 可用于在其他非handler的地方发送数据 */
     private Channel channel;
 
     public TcpClient(String host, int port) {
         this(host, port, new ExponentialBackOffRetry(1000, Integer.MAX_VALUE, 60 * 1000));
     }
 
     public TcpClient(String host, int port, RetryPolicy retryPolicy) {
         this.host = host;
         this.port = port;
         init();
     }
 
     /**
      * 向远程TCP服务器请求连接
      */
     public void connect() {
         synchronized (bootstrap) {
             ChannelFuture future = bootstrap.connect(host, port);
             this.channel = future.channel();
         }
     }
 
     private void init() {
         EventLoopGroup group = new NioEventLoopGroup();
         // bootstrap 可重用, 只需在TcpClient实例化的时候初始化即可.
         bootstrap = new Bootstrap();
         bootstrap.group(group)
                 .channel(NioSocketChannel.class)
                 .handler(new ClientHandlersInitializer(TcpClient.this));
     }
 
     public static void main(String[] args) {
         TcpClient tcpClient = new TcpClient("localhost", 2222);
         tcpClient.connect();
     }
 
 }

##### Server端 #####

###### ServerIdleStateTrigger —— 断连触发器 ######

/**
  * 在规定时间内未收到客户端的任何数据包, 将主动断开该连接
  */
 public class ServerIdleStateTrigger extends ChannelInboundHandlerAdapter {
     @Override
     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
         if (evt instanceof IdleStateEvent) {
             IdleState state = ((IdleStateEvent) evt).state();
             if (state == IdleState.READER_IDLE) {
                 // 在规定时间内没有收到客户端的上行数据, 主动断开连接
                 ctx.disconnect();
             }
         } else {
             super.userEventTriggered(ctx, evt);
         }
     }
 }

###### ServerBizHandler —— 服务器端的业务处理器 ######

/**
  * 收到来自客户端的数据包后, 直接在控制台打印出来.
  */
 @ChannelHandler.Sharable
 public class ServerBizHandler extends SimpleChannelInboundHandler<String> {
 
     private final String REC_HEART_BEAT = "I had received the heart beat!";
 
     @Override
     protected void channelRead0(ChannelHandlerContext ctx, String data) throws Exception {
         try {
             System.out.println("receive data: " + data);
 //            ctx.writeAndFlush(REC_HEART_BEAT);
         } catch (Exception e) {
             e.printStackTrace();
         }
     }
 
     @Override
     public void channelActive(ChannelHandlerContext ctx) throws Exception {
         System.out.println("Established connection with the remote client.");
 
         // do something
 
         ctx.fireChannelActive();
     }
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
         System.out.println("Disconnected with the remote client.");
 
         // do something
 
         ctx.fireChannelInactive();
     }
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
         cause.printStackTrace();
         ctx.close();
     }
 }

###### ServerHandlerInitializer —— 服务器端处理器集合的初始化类 ######

/**
  * 用于初始化服务器端涉及到的所有<code>Handler</code>
  */
 public class ServerHandlerInitializer extends ChannelInitializer<SocketChannel> {
 
     protected void initChannel(SocketChannel ch) throws Exception {
         ch.pipeline().addLast("idleStateHandler", new IdleStateHandler(5, 0, 0));
         ch.pipeline().addLast("idleStateTrigger", new ServerIdleStateTrigger());
         ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
         ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(4));
         ch.pipeline().addLast("decoder", new StringDecoder());
         ch.pipeline().addLast("encoder", new StringEncoder());
         ch.pipeline().addLast("bizHandler", new ServerBizHandler());
     }
 
 }

> 注：`new IdleStateHandler(5, 0, 0)`该`handler`代表如果在5秒内没有收到来自客户端的任何数据包（包括但不限于心跳包），将会主动断开与该客户端的连接。

###### TcpServer —— 服务器端 ######

public class TcpServer {
        private int port;
        private ServerHandlerInitializer serverHandlerInitializer;
    
        public TcpServer(int port) {
            this.port = port;
            this.serverHandlerInitializer = new ServerHandlerInitializer();
        }
    
        public void start() {
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap bootstrap = new ServerBootstrap();
                bootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(this.serverHandlerInitializer);
                // 绑定端口，开始接收进来的连接
                ChannelFuture future = bootstrap.bind(port).sync();
    
                System.out.println("Server start listen at " + port);
                future.channel().closeFuture().sync();
            } catch (Exception e) {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) throws Exception {
            int port = 2222;
            new TcpServer(port).start();
        }
    }

至此，所有代码已经编写完毕。

#### 测试 ####

首先启动客户端，再启动服务器端。启动完成后，在客户端的控制台上，可以看到打印如下类似日志：

![c1df4123d967fd05f9d92f56fd12da0b.png][]

客户端控制台输出的日志

在服务器端可以看到控制台输出了类似如下的日志：

![ed86f2cd16eeadd0317f4cdf9b7caf7c.png][]

服务器端控制台输出的日志

可以看到，客户端在发送4个心跳包后，第5个包因为等待时间较长，等到真正发送的时候，发现连接已断开了；而服务器端收到客户端的4个心跳数据包后，迟迟等不到下一个数据包，所以果断断开该连接。

在测试过程中，有可能会出现如下情况：

![3b20b1aae20aa7fa559f82716b0a0a37.png][]

异常情况

出现这种情况的原因是：在连接已断开的情况下，仍然向服务器端发送心跳包。虽然在发送心跳包之前会使用判断连接是否可用，但也有可能上一刻判断结果为可用，但下一刻发送数据包之前，连接就断了。

目前尚未找到优雅处理这种情况的方案，各位看官如果有好的解决方案，还望不吝赐教。拜谢！！！

### 断线重连 ###

断线重连这里就不过多介绍，相信各位都知道是怎么回事。这里只说大致思路，然后直接上代码。

#### 实现思路 ####

客户端在监测到与服务器端的连接断开后，或者一开始就无法连接的情况下，使用指定的重连策略进行重连操作，直到重新建立连接或重试次数耗尽。

对于如何监测连接是否断开，则是通过重写`ChannelInboundHandler#channelInactive`来实现，但连接不可用，该方法会被触发，所以只需要在该方法做好重连工作即可。

#### 代码实现 ####

> 注：以下代码都是在上面心跳机制的基础上修改/添加的。

因为断线重连是客户端的工作，所以只需对客户端代码进行修改。

#### 重试策略 ####

##### RetryPolicy —— 重试策略接口 #####

public interface RetryPolicy {
    
        /**
         * Called when an operation has failed for some reason. This method should return
         * true to make another attempt.
         *
         * @param retryCount the number of times retried so far (0 the first time)
         * @return true/false
         */
        boolean allowRetry(int retryCount);
    
        /**
         * get sleep time in ms of current retry count.
         *
         * @param retryCount current retry count
         * @return the time to sleep
         */
        long getSleepTimeMs(int retryCount);
    }

##### ExponentialBackOffRetry —— 重连策略的默认实现 #####

/**
  * Retry policy that retries a set number of times with increasing sleep time between retries
  */
 public class ExponentialBackOffRetry implements RetryPolicy {
 
     private static final int MAX_RETRIES_LIMIT = 29;
     private static final int DEFAULT_MAX_SLEEP_MS = Integer.MAX_VALUE;
 
     private final Random random = new Random();
     private final long baseSleepTimeMs;
     private final int maxRetries;
     private final int maxSleepMs;
 
     public ExponentialBackOffRetry(int baseSleepTimeMs, int maxRetries) {
         this(baseSleepTimeMs, maxRetries, DEFAULT_MAX_SLEEP_MS);
     }
 
     public ExponentialBackOffRetry(int baseSleepTimeMs, int maxRetries, int maxSleepMs) {
         this.maxRetries = maxRetries;
         this.baseSleepTimeMs = baseSleepTimeMs;
         this.maxSleepMs = maxSleepMs;
     }
 
     @Override
     public boolean allowRetry(int retryCount) {
         if (retryCount < maxRetries) {
             return true;
         }
         return false;
     }
 
     @Override
     public long getSleepTimeMs(int retryCount) {
         if (retryCount < 0) {
             throw new IllegalArgumentException("retries count must greater than 0.");
         }
         if (retryCount > MAX_RETRIES_LIMIT) {
             System.out.println(String.format("maxRetries too large (%d). Pinning to %d", maxRetries, MAX_RETRIES_LIMIT));
             retryCount = MAX_RETRIES_LIMIT;
         }
         long sleepMs = baseSleepTimeMs * Math.max(1, random.nextInt(1 << retryCount));
         if (sleepMs > maxSleepMs) {
             System.out.println(String.format("Sleep extension too large (%d). Pinning to %d", sleepMs, maxSleepMs));
             sleepMs = maxSleepMs;
         }
         return sleepMs;
     }
 }

#### ReconnectHandler—— 重连处理器 ####

@ChannelHandler.Sharable
    public class ReconnectHandler extends ChannelInboundHandlerAdapter {
    
        private int retries = 0;
        private RetryPolicy retryPolicy;
    
        private TcpClient tcpClient;
    
        public ReconnectHandler(TcpClient tcpClient) {
            this.tcpClient = tcpClient;
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("Successfully established a connection to the server.");
            retries = 0;
            ctx.fireChannelActive();
        }
    
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            if (retries == 0) {
                System.err.println("Lost the TCP connection with the server.");
                ctx.close();
            }
    
            boolean allowRetry = getRetryPolicy().allowRetry(retries);
            if (allowRetry) {
    
                long sleepTimeMs = getRetryPolicy().getSleepTimeMs(retries);
    
                System.out.println(String.format("Try to reconnect to the server after %dms. Retry count: %d.", sleepTimeMs, ++retries));
    
                final EventLoop eventLoop = ctx.channel().eventLoop();
                eventLoop.schedule(() -> {
                    System.out.println("Reconnecting ...");
                    tcpClient.connect();
                }, sleepTimeMs, TimeUnit.MILLISECONDS);
            }
            ctx.fireChannelInactive();
        }
    
    
        private RetryPolicy getRetryPolicy() {
            if (this.retryPolicy == null) {
                this.retryPolicy = tcpClient.getRetryPolicy();
            }
            return this.retryPolicy;
        }
    }

#### ClientHandlersInitializer ####

在之前的基础上，添加了重连处理器`ReconnectHandler`。

public class ClientHandlersInitializer extends ChannelInitializer<SocketChannel> {
 
     private ReconnectHandler reconnectHandler;
     private EchoHandler echoHandler;
 
     public ClientHandlersInitializer(TcpClient tcpClient) {
         Assert.notNull(tcpClient, "TcpClient can not be null.");
         this.reconnectHandler = new ReconnectHandler(tcpClient);
         this.echoHandler = new EchoHandler();
     }
 
     @Override
     protected void initChannel(SocketChannel ch) throws Exception {
         ChannelPipeline pipeline = ch.pipeline();
         pipeline.addLast(this.reconnectHandler);
         pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
         pipeline.addLast(new LengthFieldPrepender(4));
         pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
         pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
         pipeline.addLast(new Pinger());
     }
 }

#### TcpClient ####

在之前的基础上添加重连、重连策略的支持。

public class TcpClient {
 
     private String host;
     private int port;
     private Bootstrap bootstrap;
     /** 重连策略 */
     private RetryPolicy retryPolicy;
     /** 将<code>Channel</code>保存起来, 可用于在其他非handler的地方发送数据 */
     private Channel channel;
 
     public TcpClient(String host, int port) {
         this(host, port, new ExponentialBackOffRetry(1000, Integer.MAX_VALUE, 60 * 1000));
     }
 
     public TcpClient(String host, int port, RetryPolicy retryPolicy) {
         this.host = host;
         this.port = port;
         this.retryPolicy = retryPolicy;
         init();
     }
 
     /**
      * 向远程TCP服务器请求连接
      */
     public void connect() {
         synchronized (bootstrap) {
             ChannelFuture future = bootstrap.connect(host, port);
             future.addListener(getConnectionListener());
             this.channel = future.channel();
         }
     }
 
     public RetryPolicy getRetryPolicy() {
         return retryPolicy;
     }
 
     private void init() {
         EventLoopGroup group = new NioEventLoopGroup();
         // bootstrap 可重用, 只需在TcpClient实例化的时候初始化即可.
         bootstrap = new Bootstrap();
         bootstrap.group(group)
                 .channel(NioSocketChannel.class)
                 .handler(new ClientHandlersInitializer(TcpClient.this));
     }
 
     private ChannelFutureListener getConnectionListener() {
         return new ChannelFutureListener() {
             @Override
             public void operationComplete(ChannelFuture future) throws Exception {
                 if (!future.isSuccess()) {
                     future.channel().pipeline().fireChannelInactive();
                 }
             }
         };
     }
 
     public static void main(String[] args) {
         TcpClient tcpClient = new TcpClient("localhost", 2222);
         tcpClient.connect();
     }
 
 }

### 测试 ###

在测试之前，为了避开 `Connection reset by peer` 异常，可以稍微修改`Pinger`的`ping()`方法，添加`if (second == 5)`的条件判断。如下：

private void ping(Channel channel) {
         int second = Math.max(1, random.nextInt(baseRandom));
         if (second == 5) {
             second = 6;
         }
         System.out.println("next heart beat will send after " + second + "s.");
         ScheduledFuture<?> future = channel.eventLoop().schedule(new Runnable() {
             @Override
             public void run() {
                 if (channel.isActive()) {
                     System.out.println("sending heart beat to the server...");
                     channel.writeAndFlush(ClientIdleStateTrigger.HEART_BEAT);
                 } else {
                     System.err.println("The connection had broken, cancel the task that will send a heart beat.");
                     channel.closeFuture();
                     throw new RuntimeException();
                 }
             }
         }, second, TimeUnit.SECONDS);
 
         future.addListener(new GenericFutureListener() {
             @Override
             public void operationComplete(Future future) throws Exception {
                 if (future.isSuccess()) {
                     ping(channel);
                 }
             }
         });
     }

#### 启动客户端 ####

先只启动客户端，观察控制台输出，可以看到类似如下日志：

![9a291d5306eef79fcca23b0cdafceede.png][]

断线重连测试——客户端控制台输出

可以看到，当客户端发现无法连接到服务器端，所以一直尝试重连。随着重试次数增加，重试时间间隔越大，但又不想无限增大下去，所以需要定一个阈值，比如60s。如上图所示，当下一次重试时间超过60s时，会打印`Sleep extension too large(*). Pinning to 60000`，单位为ms。出现这句话的意思是，计算出来的时间超过阈值（60s），所以把真正睡眠的时间重置为阈值（60s）。

#### 启动服务器端 ####

接着启动服务器端，然后继续观察客户端控制台输出。

![dd48288dc6537de177c027eddedaa5f4.png][] 图片

断线重连测试——服务器端启动后客户端控制台输出

可以看到，在第9次重试失败后，第10次重试之前，启动的服务器，所以第10次重连的结果为，即成功连接到服务器。接下来因为还是不定时服务器，所以出现断线重连、断线重连的循环。

#### 扩展 ####

在不同环境，可能会有不同的重连需求。有不同的重连需求的，只需自己实现`RetryPolicy`接口，然后在创建`TcpClient`的时候覆盖默认的重连策略即可。

--------------------

[c1df4123d967fd05f9d92f56fd12da0b.png]: https://img-blog.csdnimg.cn/img_convert/c1df4123d967fd05f9d92f56fd12da0b.png
[ed86f2cd16eeadd0317f4cdf9b7caf7c.png]: https://img-blog.csdnimg.cn/img_convert/ed86f2cd16eeadd0317f4cdf9b7caf7c.png
[3b20b1aae20aa7fa559f82716b0a0a37.png]: https://img-blog.csdnimg.cn/img_convert/3b20b1aae20aa7fa559f82716b0a0a37.png
[9a291d5306eef79fcca23b0cdafceede.png]: https://img-blog.csdnimg.cn/img_convert/9a291d5306eef79fcca23b0cdafceede.png
[dd48288dc6537de177c027eddedaa5f4.png]: https://img-blog.csdnimg.cn/img_convert/dd48288dc6537de177c027eddedaa5f4.png