更多内容，前往个人博客
所谓心跳，即在 TCP 长连接中, 客户端和服务器之间定期发送的一种特殊的数据包，通知对方自己还在线，以确保 TCP 连接的有效性。心跳包还有另一个作用，经常被忽略，即：一个连接如果长时间不用，防火墙或者路由器就会断开该连接。建议：将下面的代码敲一遍，对这个流程就有一个比较好的理解。
一、核心Handler
在 Netty 中，实现心跳机制的关键是 IdleStateHandler，那么这个 Handler 如何使用呢? 先看下它的构造器：
1 public IdleStateHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) {
2 this((long)readerIdleTimeSeconds, (long)writerIdleTimeSeconds, (long)allIdleTimeSeconds, TimeUnit.SECONDS);
3 }
这里解释下三个参数的含义：
【1】readerIdleTimeSeconds: 读超时。即当在指定的时间间隔内没有从 Channel 读取到数据时, 会触发一个 READER_IDLE 的 IdleStateEvent 事件。
【2】writerIdleTimeSeconds: 写超时。即当在指定的时间间隔内没有数据写入到 Channel 时, 会触发一个 WRITER_IDLE 的 IdleStateEvent 事件。
【3】allIdleTimeSeconds: 读/写超时。即当在指定的时间间隔内没有读或写操作时，会触发一个 ALL_IDLE 的 IdleStateEvent 事件。
这三个参数默认的时间单位是秒。若需要指定其他时间单位，可以使用另一个构造方法：
IdleStateHandler(boolean observeOutput, long readerIdleTime, long writerIdleTime, long allIdleTime, TimeUnit unit)
IdleStateHandler 的实现原理。
下面将使用
IdleStateHandler来实现心跳，Client端连接到Server端后，会循环执行一个任务：随机等待几秒，然后ping一下Server端，即发送一个心跳包。当等待的时间超过规定时间，将会发送失败，以为Server端在此之前已经主动断开连接了。
Client端
ClientIdleStateTrigger（心跳触发器）类 ClientIdleStateTrigger也是一个 Handler，只是重写了userEventTriggered方法，用于捕获 IdleState.WRITER_IDLE事件（未在指定时间内向服务器发送数据），然后向 Server端发送一个心跳包。
1 package com.yunda.netty;
2
3 import io.netty.buffer.ByteBuf;
4 import io.netty.buffer.Unpooled;
5 import io.netty.channel.ChannelHandler;
6 import io.netty.channel.ChannelHandlerContext;
7 import io.netty.channel.ChannelInboundHandlerAdapter;
8 import io.netty.handler.timeout.IdleState;
9 import io.netty.handler.timeout.IdleStateEvent;
10 import io.netty.util.CharsetUtil;
11
12 /**
13 * @description: 用于捕获{@link IdleState#WRITER_IDLE}事件（未在指定时间内向服务器发送数据），然后向Server端发送一个心跳包。
14 * @author: zzx
15 * @createDate: 2020/9/27
16 * @version: 1.0
17 */
18 @ChannelHandler.Sharable
19 public class ConnectorIdleStateTrigger extends ChannelInboundHandlerAdapter {
20
21 private static final ByteBuf HEARTBEAT_SEQUENCE = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Heartbeat",
22 CharsetUtil.UTF_8));
23
24 @Override
25 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
26 if (evt instanceof IdleStateEvent) {
27 IdleState state = ((IdleStateEvent) evt).state();
28 if (state == IdleState.WRITER_IDLE) {
29 // write heartbeat to server
30 ctx.writeAndFlush(HEARTBEAT_SEQUENCE.duplicate());
31 }
32 } else {
33 super.userEventTriggered(ctx, evt);
34 }
35 }
36 }
接下来就是重点，我们需要写一个类，这个类可以观察链路是否断了，如果断了，进行循环的断线重连操作，ConnectionWatch，链路检测，完整代码如下：
1 package com.yunda.netty.adml;
2
3 import io.netty.bootstrap.Bootstrap;
4 import io.netty.channel.*;
5 import io.netty.util.Timeout;
6 import io.netty.util.Timer;
7 import io.netty.util.TimerTask;
8 import org.slf4j.Logger;
9 import org.slf4j.LoggerFactory;
10
11 import java.util.concurrent.TimeUnit;
12
13 /**
14 * @description:重连检测，当发现当前的链路不稳定关闭之后，进行12次重连
15 * @author: zzx
16 * @createDate: 2020/9/27
17 * @version: 1.0
18 */
19 @ChannelHandler.Sharable
20 public class ConnectionWatch extends ChannelInboundHandlerAdapter implements TimerTask{
21 /**
22 * 日志
23 */
24 private static final Logger logger= LoggerFactory.getLogger(SubClientBootStart.class);
25
26 /**
27 * 建立连接的 Bootstrap 从 TCPClinet中获取
28 */
29 private final Bootstrap bootstrap;
30
31 /**
32 * JDK 自带定时任务类
33 */
34 private final Timer timer;
35
36 /**
37 * 端口，从 TCPClinet中获取
38 */
39 private final int port;
40
41 /**
42 * ip，从 TCPClinet中获取
43 */
44 private final String host;
45
46 /**
47 * 是否重连
48 */
49 private volatile boolean reconnect = true;
50
51 /**
52 * 重连次数
53 */
54 private int attempts;
55
56
57 /**
58 * 有参构造器
59 */
60 public ConnectionWatch(Bootstrap bootstrap, Timer timer, int port, String host, boolean reconnect) {
61 this.bootstrap = bootstrap;
62 this.timer = timer;
63 this.port = port;
64 this.host = host;
65 this.reconnect = reconnect;
66 }
67
68 /**
69 * channel链路每次active的时候，将其连接的次数重新☞ 0
70 */
71 @Override
72 public void channelActive(ChannelHandlerContext ctx) throws Exception {
73 logger.info("当前链路已经激活了，重连尝试次数重新置为0");
74 attempts = 0;
75 ctx.fireChannelActive();
76 }
77
78 /**
79 * 当断开连接时调用该方法
80 */
81 @Override
82 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
83 if(reconnect){
84 logger.info("链接关闭，将进行重连");
85 if (attempts < 12) {
86 attempts++;
87 //重连的间隔时间会越来越长
88 int timeout = 2 << attempts;
89 /**创建一个定时任务执行重连操作*/
90 timer.newTimeout(this, timeout, TimeUnit.MILLISECONDS);
91 }
92 }
93 ctx.fireChannelInactive();
94 }
95
96 /**
97 * 定时任务执行的方法
98 */
99 @Override
100 public void run(Timeout timeout) throws Exception {
101
102 ChannelFuture future;
103 //bootstrap已经初始化好了，只需要将handler填入就可以了
104 synchronized (bootstrap) {
105 bootstrap.handler(new ChannelInitializer<Channel>() {
106
107 @Override
108 protected void initChannel(Channel ch) throws Exception {
109 ch.pipeline().addLast(handlers());
110 }
111 });
112 future = bootstrap.connect(host,port);
113 }
114 //future对象
115 future.addListener(new ChannelFutureListener() {
116 @Override
117 public void operationComplete(ChannelFuture f) throws Exception {
118 boolean succeed = f.isSuccess();
119
120 //如果重连失败，则调用ChannelInactive方法，再次出发重连事件，一直尝试12次，如果失败则不再重连
121 if (!succeed) {
122 logger.info("重连失败");
123 f.channel().pipeline().fireChannelInactive();
124 }else{
125 logger.info("重连成功");
126 }
127 }
128 });
129 }
130
131 /**
132 * 钩子方法 ， 存放所有的 handler类
133 */
134 public ChannelHandler[] handlers() {
135 return null;
136 }
137 }
HeartBeatClientHandler：客户端注入的 handler业务处理类，主要用作业务处理。
1 package com.yunda.netty;
2
3 import io.netty.channel.ChannelHandler;
4 import io.netty.channel.ChannelHandlerContext;
5 import io.netty.channel.ChannelInboundHandlerAdapter;
6 import io.netty.util.ReferenceCountUtil;
7
8 import java.util.Date;
9
10 /**
11 * @description: 客户端注入的handler类
12 * @author: zzx
13 * @createDate: 2020/9/27
14 * @version: 1.0
15 */
16 @ChannelHandler.Sharable
17 public class HeartBeatClientHandler extends ChannelInboundHandlerAdapter {
18
19
20 @Override
21 public void channelActive(ChannelHandlerContext ctx) throws Exception {
22 System.out.println("激活时间是："+new Date());
23 System.out.println("HeartBeatClientHandler channelActive");
24 ctx.fireChannelActive();
25 }
26
27 @Override
28 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
29 System.out.println("停止时间是："+new Date());
30 System.out.println("HeartBeatClientHandler channelInactive");
31 }
32
33
34 @Override
35 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
36 String message = (String) msg;
37 System.out.println(message);
38 if (message.equals("Heartbeat")) {
39 ctx.write("has read message from server");
40 ctx.flush();
41 }
42 ReferenceCountUtil.release(msg);
43 }
44 }
HeartBeatsClient：TCP连接的客户端
1 package com.yunda.netty;
2
3 import com.yunda.netty.adml.ConnectionWatch;
4 import com.yunda.netty.adml.ConnectorIdleStateTrigger;
5 import io.netty.bootstrap.Bootstrap;
6 import io.netty.channel.*;
7 import io.netty.channel.nio.NioEventLoopGroup;
8 import io.netty.channel.socket.nio.NioSocketChannel;
9 import io.netty.handler.codec.string.StringDecoder;
10 import io.netty.handler.codec.string.StringEncoder;
11 import io.netty.handler.logging.LogLevel;
12 import io.netty.handler.logging.LoggingHandler;
13 import io.netty.handler.timeout.IdleStateHandler;
14 import io.netty.util.HashedWheelTimer;
15
16 import java.util.concurrent.TimeUnit;
17
18 /**
19 * @description:
20 * @author: zzx
21 * @createDate: 2020/9/27
22 * @version: 1.0
23 */
24 public class HeartBeatsClient {
25
26 /**
27 * 定时任务具体实现类
28 */
29 protected final HashedWheelTimer timer = new HashedWheelTimer();
30
31 /**
32 * 客户端启动 Bootstrap
33 */
34 private Bootstrap bootstrap;
35
36 /**
37 * 心跳检测触发器
38 */
39 private final ConnectorIdleStateTrigger idleStateTrigger = new ConnectorIdleStateTrigger();
40
41 /**
42 * 连接 服务端
43 * @param port 端口
44 * @param host ip
45 * @throws Exception
46 */
47 public void connect(int port, String host) throws Exception {
48 //NIO 非阻塞循环组
49 EventLoopGroup group = new NioEventLoopGroup();
50 //客户端 Bootstrap
51 bootstrap = new Bootstrap();
52 //添加组，线程类型，日志handler
53 bootstrap.group(group).channel(NioSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO));
54
55 //新建 监控类，将连接信息都传输过去
56 final ConnectionWatch watchdog = new ConnectionWatch(bootstrap, timer, port,host, true) {
57 @Override
58 public ChannelHandler[] handlers() {
59 return new ChannelHandler[] {
60 this,
61 new IdleStateHandler(4, 4, 4, TimeUnit.SECONDS),
62 idleStateTrigger,
63 new StringDecoder(),
64 new StringEncoder(),
65 new HeartBeatClientHandler()
66 };
67 }
68 };
69
70 //连接后的返回类
71 ChannelFuture future;
72 /**
73 * 开始连接 初始化handler职责链
74 */
75 try {
76 synchronized (bootstrap) {
77 bootstrap.handler(new ChannelInitializer<Channel>() {
78 //初始化channel
79 @Override
80 protected void initChannel(Channel ch) throws Exception {
81 ch.pipeline().addLast(watchdog.handlers());
82 }
83 });
84 //建立连接
85 future = bootstrap.connect(host,port);
86 }
87
88 // 以上代码在synchronized同步块外面是安全的
89 future.sync();
90 } catch (Throwable t) {
91 throw new Exception("connects to fails", t);
92 }
93 }
94
95 /**
96 * @param args
97 * @throws Exception
98 */
99 public static void main(String[] args) throws Exception {
100 int port = 1123;
101 if (args != null && args.length > 0) {
102 try {
103 port = Integer.valueOf(args[0]);
104 } catch (NumberFormatException e) {
105 // 采用默认值
106 }
107 }
108 new HeartBeatsClient().connect(port, "127.0.0.1");
109 }
110 }
Server端
AcceptorIdleStateTrigger：断连触发器
1 package com.dahua.netty.zzx;
2
3 import io.netty.channel.ChannelHandlerContext;
4 import io.netty.channel.ChannelInboundHandlerAdapter;
5 import io.netty.handler.timeout.IdleState;
6 import io.netty.handler.timeout.IdleStateEvent;
7
8 /**
9 * @description: 在规定时间内未收到客户端的任何数据包, 将主动断开该连接
10 * @author: zzx
11 * @createDate: 2020/9/27
12 * @version: 1.0
13 */
14 public class AcceptorIdleStateTrigger extends ChannelInboundHandlerAdapter {
15 @Override
16 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
17 if (evt instanceof IdleStateEvent) {
18 IdleState state = ((IdleStateEvent) evt).state();
19 if (state == IdleState.READER_IDLE) {
20 throw new Exception("idle exception");
21 }
22 } else {
23 super.userEventTriggered(ctx, evt);
24 }
25 }
26 }
HeartBeatServerHandler：服务器端的业务处理器
1 package com.dahua.netty.zzx;
2
3 import io.netty.channel.ChannelHandlerContext;
4 import io.netty.channel.ChannelInboundHandlerAdapter;
5
6 /**
7 * @description: 打印客户端发送的消息
8 * @author: zzx
9 * @createDate: 2020/9/27
10 * @version: 1.0
11 */
12 public class HeartBeatServerHandler extends ChannelInboundHandlerAdapter {
13 @Override
14 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
15 System.out.println("server channelRead..");
16 System.out.println(ctx.channel().remoteAddress() + "->Server :" + msg.toString());
17 }
18
19 @Override
20 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
21 cause.printStackTrace();
22 ctx.close();
23 }
24 }
TestBootStrap：服务器端，new IdleStateHandler(5, 0, 0)该handler代表如果在5秒内没有收到来自客户端的任何数据包（包括但不限于心跳包），将会主动断开与该客户端的连接。
1 package com.dahua.netty.zzx;
2
3 import com.dahua.netty.NettyServer;
4 import com.dahua.netty.zzx.AcceptorIdleStateTrigger;
5 import com.dahua.netty.zzx.HeartBeatServerHandler;
6 import io.netty.channel.ChannelHandlerAdapter;
7 import io.netty.handler.codec.string.StringDecoder;
8 import io.netty.handler.codec.string.StringEncoder;
9 import io.netty.handler.timeout.IdleStateHandler;
10
11 import java.nio.charset.Charset;
12 import java.util.LinkedHashMap;
13 import java.util.concurrent.TimeUnit;
14
15 /**
16 * @description:
17 * @author: zzx
18 * @createDate: 2020/9/16
19 * @version: 1.0
20 */
21 public class TestBootStrap {
22 public static void main(String[] args) {
23 /*-------------------------------测试代码，建立服务端---------------------------------*/
24 //子系统接口对接,端口使用
25 NettyServer nettyServer = new NettyServer(1123);
26 LinkedHashMap<String, ChannelHandlerAdapter> subChannelHandlerAdapterHashMap =new LinkedHashMap<>();
27 subChannelHandlerAdapterHashMap.put("idleStateHandler",new IdleStateHandler(5, 5, 5, TimeUnit.SECONDS));
28 subChannelHandlerAdapterHashMap.put("acceptorIdleStateTrigger",new AcceptorIdleStateTrigger());
29 subChannelHandlerAdapterHashMap.put("stringDecoder",new StringDecoder(Charset.forName("utf-8")));
30 subChannelHandlerAdapterHashMap.put("stringEncoder",new StringEncoder(Charset.forName("utf-8")));
31 subChannelHandlerAdapterHashMap.put("heartBeatServerHandler",new HeartBeatServerHandler());
32
33 try {
34 nettyServer.start(subChannelHandlerAdapterHashMap);
35 } catch (Exception e) {
36 e.printStackTrace();
37 }
38 /*----------------------------------测试代码结束-------------------------------------------*/
39 }
40 }
测试
首先启动客户端，再启动服务器端。启动完成后，在客户端的控制台上，可以看到打印如下类似日志：
在服务器端可以看到控制台输出了类似如下的日志：
可以看到，客户端在发送4个心跳包后，第5个包因为等待时间较长，等到真正发送的时候，发现连接已断开了；而服务器端收到客户端的4个心跳数据包后，迟迟等不到下一个数据包，所以果断断开该连接。
异常情况
在测试过程中，有可能会出现如下情况：
出现这种情况的原因是：在连接已断开的情况下，仍然向服务器端发送心跳包。虽然在发送心跳包之前会使用channel.isActive()判断连接是否可用，但也有可能上一刻判断结果为可用，但下一刻发送数据包之前，连接就断了。目前尚未找到优雅处理这种情况的方案。
二、断线重连
实现思路：客户端在监测到与服务器端的连接断开后，或者一开始就无法连接的情况下，使用指定的重连策略进行重连操作，直到重新建立连接或重试次数耗尽。对于如何监测连接是否断开，则是通过重写 ChannelInboundHandler#channelInactive来实现，当连接不可用，该方法会被触发，所以只需要在该方法做好重连工作即可。
代码实现：以下代码都是在上面心跳机制的基础上修改/添加的。因为断线重连是客户端的工作，所以只需对客户端代码进行修改。
【1】RetryPolicy重试策略接口
1 public interface RetryPolicy {
2
3 /**
4 * Called when an operation has failed for some reason. This method should return
5 * true to make another attempt.
6 *
7 * @param retryCount the number of times retried so far (0 the first time)
8 * @return true/false
9 */
10 boolean allowRetry(int retryCount);
11
12 /**
13 * get sleep time in ms of current retry count.
14 *
15 * @param retryCount current retry count
16 * @return the time to sleep
17 */
18 long getSleepTimeMs(int retryCount);
19 }
【2】ExponentialBackOffRetry重连策略的默认实现
1 /**
2 * <p>Retry policy that retries a set number of times with increasing sleep time between retries</p>
3 */
4 public class ExponentialBackOffRetry implements RetryPolicy {
5
6 private static final int MAX_RETRIES_LIMIT = 29;
7 private static final int DEFAULT_MAX_SLEEP_MS = Integer.MAX_VALUE;
8
9 private final Random random = new Random();
10 private final long baseSleepTimeMs;
11 private final int maxRetries;
12 private final int maxSleepMs;
13
14 public ExponentialBackOffRetry(int baseSleepTimeMs, int maxRetries) {
15 this(baseSleepTimeMs, maxRetries, DEFAULT_MAX_SLEEP_MS);
16 }
17
18 public ExponentialBackOffRetry(int baseSleepTimeMs, int maxRetries, int maxSleepMs) {
19 this.maxRetries = maxRetries;
20 this.baseSleepTimeMs = baseSleepTimeMs;
21 this.maxSleepMs = maxSleepMs;
22 }
23
24 @Override
25 public boolean allowRetry(int retryCount) {
26 if (retryCount < maxRetries) {
27 return true;
28 }
29 return false;
30 }
31
32 @Override
33 public long getSleepTimeMs(int retryCount) {
34 if (retryCount < 0) {
35 throw new IllegalArgumentException("retries count must greater than 0.");
36 }
37 if (retryCount > MAX_RETRIES_LIMIT) {
38 System.out.println(String.format("maxRetries too large (%d). Pinning to %d", maxRetries, MAX_RETRIES_LIMIT));
39 retryCount = MAX_RETRIES_LIMIT;
40 }
41 long sleepMs = baseSleepTimeMs * Math.max(1, random.nextInt(1 << retryCount));
42 if (sleepMs > maxSleepMs) {
43 System.out.println(String.format("Sleep extension too large (%d). Pinning to %d", sleepMs, maxSleepMs));
44 sleepMs = maxSleepMs;
45 }
46 return sleepMs;
47 }
48 }
【3】ReconnectHandler重连处理器
1 @ChannelHandler.Sharable
2 public class ReconnectHandler extends ChannelInboundHandlerAdapter {
3
4 private int retries = 0;
5 private RetryPolicy retryPolicy;
6
7 private TcpClient tcpClient;
8
9 public ReconnectHandler(TcpClient tcpClient) {
10 this.tcpClient = tcpClient;
11 }
12
13 @Override
14 public void channelActive(ChannelHandlerContext ctx) throws Exception {
15 System.out.println("Successfully established a connection to the server.");
16 retries = 0;
17 ctx.fireChannelActive();
18 }
19
20 @Override
21 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
22 if (retries == 0) {
23 System.err.println("Lost the TCP connection with the server.");
24 ctx.close();
25 }
26
27 boolean allowRetry = getRetryPolicy().allowRetry(retries);
28 if (allowRetry) {
29
30 long sleepTimeMs = getRetryPolicy().getSleepTimeMs(retries);
31
32 System.out.println(String.format("Try to reconnect to the server after %dms. Retry count: %d.", sleepTimeMs, ++retries));
33
34 final EventLoop eventLoop = ctx.channel().eventLoop();
35 eventLoop.schedule(() -> {
36 System.out.println("Reconnecting ...");
37 tcpClient.connect();
38 }, sleepTimeMs, TimeUnit.MILLISECONDS);
39 }
40 ctx.fireChannelInactive();
41 }
42
43
44 private RetryPolicy getRetryPolicy() {
45 if (this.retryPolicy == null) {
46 this.retryPolicy = tcpClient.getRetryPolicy();
47 }
48 return this.retryPolicy;
49 }
50 }
【4】ClientHandlersInitializer: 在之前的基础上，添加了重连处理器ReconnectHandler。
1 public class ClientHandlersInitializer extends ChannelInitializer<SocketChannel> {
2
3 private ReconnectHandler reconnectHandler;
4 private EchoHandler echoHandler;
5
6 public ClientHandlersInitializer(TcpClient tcpClient) {
7 Assert.notNull(tcpClient, "TcpClient can not be null.");
8 this.reconnectHandler = new ReconnectHandler(tcpClient);
9 this.echoHandler = new EchoHandler();
10 }
11
12 @Override
13 protected void initChannel(SocketChannel ch) throws Exception {
14 ChannelPipeline pipeline = ch.pipeline();
15 pipeline.addLast(this.reconnectHandler);
16 pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
17 pipeline.addLast(new LengthFieldPrepender(4));
18 pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
19 pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
20 pipeline.addLast(new Pinger());
21 }
22 }
【5】TcpClient：在之前的基础上添加重连、重连策略的支持。
1 public class TcpClient {
2
3 private String host;
4 private int port;
5 private Bootstrap bootstrap;
6 /** 重连策略 */
7 private RetryPolicy retryPolicy;
8 /** 将<code>Channel</code>保存起来, 可用于在其他非handler的地方发送数据 */
9 private Channel channel;
10
11 public TcpClient(String host, int port) {
12 this(host, port, new ExponentialBackOffRetry(1000, Integer.MAX_VALUE, 60 * 1000));
13 }
14
15 public TcpClient(String host, int port, RetryPolicy retryPolicy) {
16 this.host = host;
17 this.port = port;
18 this.retryPolicy = retryPolicy;
19 init();
20 }
21
22 /**
23 * 向远程TCP服务器请求连接
24 */
25 public void connect() {
26 synchronized (bootstrap) {
27 ChannelFuture future = bootstrap.connect(host, port);
28 future.addListener(getConnectionListener());
29 this.channel = future.channel();
30 }
31 }
32
33 public RetryPolicy getRetryPolicy() {
34 return retryPolicy;
35 }
36
37 private void init() {
38 EventLoopGroup group = new NioEventLoopGroup();
39 // bootstrap 可重用, 只需在TcpClient实例化的时候初始化即可.
40 bootstrap = new Bootstrap();
41 bootstrap.group(group)
42 .channel(NioSocketChannel.class)
43 .handler(new ClientHandlersInitializer(TcpClient.this));
44 }
45
46 private ChannelFutureListener getConnectionListener() {
47 return new ChannelFutureListener() {
48 @Override
49 public void operationComplete(ChannelFuture future) throws Exception {
50 if (!future.isSuccess()) {
51 future.channel().pipeline().fireChannelInactive();
52 }
53 }
54 };
55 }
56
57 public static void main(String[] args) {
58 TcpClient tcpClient = new TcpClient("localhost", 2222);
59 tcpClient.connect();
60 }
61
62 }
【6】测试：在测试之前，为了避开 Connection reset by peer 异常，可以稍微修改Pinger的ping()方法，添加if (second == 5)的条件判断。如下：
1 private void ping(Channel channel) {
2 int second = Math.max(1, random.nextInt(baseRandom));
3 if (second == 5) {
4 second = 6;
5 }
6 System.out.println("next heart beat will send after " + second + "s.");
7 ScheduledFuture<?> future = channel.eventLoop().schedule(new Runnable() {
8 @Override
9 public void run() {
10 if (channel.isActive()) {
11 System.out.println("sending heart beat to the server...");
12 channel.writeAndFlush(ClientIdleStateTrigger.HEART_BEAT);
13 } else {
14 System.err.println("The connection had broken, cancel the task that will send a heart beat.");
15 channel.closeFuture();
16 throw new RuntimeException();
17 }
18 }
19 }, second, TimeUnit.SECONDS);
20
21 future.addListener(new GenericFutureListener() {
22 @Override
23 public void operationComplete(Future future) throws Exception {
24 if (future.isSuccess()) {
25 ping(channel);
26 }
27 }
28 });
29 }
启动客户端
先只启动客户端，观察控制台输出，可以看到类似如下日志： 客户端控制台输出
可以看到，当客户端发现无法连接到服务器端，所以一直尝试重连。随着重试次数增加，重试时间间隔越大，但又不想无限增大下去，所以需要定一个阈值，比如60s。如上图所示，当下一次重试时间超过60s时，会打印`Sleep extension too large(*). Pinning to 60000`，单位为ms。出现这句话的意思是，计算出来的时间超过阈值（60s），所以把真正睡眠的时间重置为阈值（60s）。
启动服务器端
接着启动服务器端，服务器端启动后客户端控制台输出。然后继续观察客户端控制台输出。
可以看到，在第9次重试失败后，第10次重试之前，启动的服务器，所以第10次重连的结果为`Successfully established a connection to the server.`，即成功连接到服务器。接下来因为还是不定时ping服务器，所以出现断线重连、断线重连的循环。
在不同环境，可能会有不同的重连需求。有不同的重连需求的，只需自己实现RetryPolicy接口，然后在创建`TcpClient`的时候覆盖默认的重连策略即可。
