点击上方蓝色“方志朋”，选择“设为星标”
回复“666”获取独家整理的学习资料！
来源：/p/1a28e48edd92
心跳机制
何为心跳
所谓心跳, 即在TCP长连接中, 客户端和服务器之间定期发送的一种特殊的数据包, 通知对方自己还在线, 以确保TCP连接的有效性.
注：心跳包还有另一个作用，经常被忽略，即：一个连接如果长时间不用，防火墙或者路由器就会断开该连接。
如何实现
核心Handler —— IdleStateHandler
在Netty中, 实现心跳机制的关键是IdleStateHandler, 那么这个Handler如何使用呢? 先看下它的构造器：
publicIdleStateHandler(intreaderIdleTimeSeconds,intwriterIdleTimeSeconds,intallIdleTimeSeconds){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端发送一个心跳包。
/***<p>*用于捕获{@linkIdleState#WRITER_IDLE}事件（未在指定时间内向服务器发送数据），然后向<code>Server</code>端发送一个心跳包。*</p>*/publicclassClientIdleStateTriggerextendsChannelInboundHandlerAdapter{publicstaticfinalStringHEART_BEAT="heartbeat!";@OverridepublicvoiduserEventTriggered(ChannelHandlerContextctx,Objectevt)throwsException{if(evtinstanceofIdleStateEvent){IdleStatestate=((IdleStateEvent)evt).state();if(state==IdleState.WRITER_IDLE){//writeheartbeattoserverctx.writeAndFlush(HEART_BEAT);}}else{super.userEventTriggered(ctx,evt);}}}
Pinger —— 心跳发射器
/***<p>客户端连接到服务器端后，会循环执行一个任务：随机等待几秒，然后ping一下Server端，即发送一个心跳包。</p>*/publicclassPingerextendsChannelInboundHandlerAdapter{privateRandomrandom=newRandom();privateintbaseRandom=8;privateChannelchannel;@OverridepublicvoidchannelActive(ChannelHandlerContextctx)throwsException{super.channelActive(ctx);this.channel=ctx.channel();ping(ctx.channel());}privatevoidping(Channelchannel){intsecond=Math.max(1,random.nextInt(baseRandom));System.out.println("nextheartbeatwillsendafter"+second+"s.");ScheduledFuture<?>future=channel.eventLoop().schedule(newRunnable(){@Overridepublicvoidrun(){if(channel.isActive()){System.out.println("sendingheartbeattotheserver...");channel.writeAndFlush(ClientIdleStateTrigger.HEART_BEAT);}else{System.err.println("Theconnectionhadbroken,cancelthetaskthatwillsendaheartbeat.");channel.closeFuture();thrownewRuntimeException();}}},second,TimeUnit.SECONDS);future.addListener(newGenericFutureListener(){@OverridepublicvoidoperationComplete(Futurefuture)throwsException{if(future.isSuccess()){ping(channel);}}});}@OverridepublicvoidexceptionCaught(ChannelHandlerContextctx,Throwablecause)throwsException{//当Channel已经断开的情况下,仍然发送数据,会抛异常,该方法会被调用.cause.printStackTrace();ctx.close();}}
ClientHandlersInitializer —— 客户端处理器集合的初始化类
publicclassClientHandlersInitializerextendsChannelInitializer<SocketChannel>{privateReconnectHandlerreconnectHandler;privateEchoHandlerechoHandler;publicClientHandlersInitializer(TcpClienttcpClient){Assert.notNull(tcpClient,"TcpClientcannotbenull.");this.reconnectHandler=newReconnectHandler(tcpClient);this.echoHandler=newEchoHandler();}@OverrideprotectedvoidinitChannel(SocketChannelch)throwsException{ChannelPipelinepipeline=ch.pipeline();pipeline.addLast(newLengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));pipeline.addLast(newLengthFieldPrepender(4));pipeline.addLast(newStringDecoder(CharsetUtil.UTF_8));pipeline.addLast(newStringEncoder(CharsetUtil.UTF_8));pipeline.addLast(newPinger());}}
注：上面的
Handler集合，除了Pinger，其他都是编解码器和解决粘包，可以忽略。
TcpClient —— TCP连接的客户端
publicclassTcpClient{privateStringhost;privateintport;privateBootstrapbootstrap;/**将<code>Channel</code>保存起来,可用于在其他非handler的地方发送数据*/privateChannelchannel;publicTcpClient(Stringhost,intport){this(host,port,newExponentialBackOffRetry(1000,Integer.MAX_VALUE,60*1000));}publicTcpClient(Stringhost,intport,RetryPolicyretryPolicy){this.host=host;this.port=port;init();}/***向远程TCP服务器请求连接*/publicvoidconnect(){synchronized(bootstrap){ChannelFuturefuture=bootstrap.connect(host,port);this.channel=future.channel();}}privatevoidinit(){EventLoopGroupgroup=newNioEventLoopGroup();//bootstrap可重用,只需在TcpClient实例化的时候初始化即可.bootstrap=newBootstrap();bootstrap.group(group).channel(NioSocketChannel.class).handler(newClientHandlersInitializer(TcpClient.this));}publicstaticvoidmain(String[]args){TcpClienttcpClient=newTcpClient("localhost",2222);tcpClient.connect();}}
Server端
ServerIdleStateTrigger —— 断连触发器
/***<p>在规定时间内未收到客户端的任何数据包,将主动断开该连接</p>*/publicclassServerIdleStateTriggerextendsChannelInboundHandlerAdapter{@OverridepublicvoiduserEventTriggered(ChannelHandlerContextctx,Objectevt)throwsException{if(evtinstanceofIdleStateEvent){IdleStatestate=((IdleStateEvent)evt).state();if(state==IdleState.READER_IDLE){//在规定时间内没有收到客户端的上行数据,主动断开连接ctx.disconnect();}}else{super.userEventTriggered(ctx,evt);}}}
ServerBizHandler —— 服务器端的业务处理器
/***<p>收到来自客户端的数据包后,直接在控制台打印出来.</p>*/@ChannelHandler.SharablepublicclassServerBizHandlerextendsSimpleChannelInboundHandler<String>{privatefinalStringREC_HEART_BEAT="Ihadreceivedtheheartbeat!";@OverrideprotectedvoidchannelRead0(ChannelHandlerContextctx,Stringdata)throwsException{try{System.out.println("receivedata:"+data);//ctx.writeAndFlush(REC_HEART_BEAT);}catch(Exceptione){e.printStackTrace();}}@OverridepublicvoidchannelActive(ChannelHandlerContextctx)throwsException{System.out.println("Establishedconnectionwiththeremoteclient.");//dosomethingctx.fireChannelActive();}@OverridepublicvoidchannelInactive(ChannelHandlerContextctx)throwsException{System.out.println("Disconnectedwiththeremoteclient.");//dosomethingctx.fireChannelInactive();}@OverridepublicvoidexceptionCaught(ChannelHandlerContextctx,Throwablecause)throwsException{cause.printStackTrace();ctx.close();}}
ServerHandlerInitializer —— 服务器端处理器集合的初始化类
/***<p>用于初始化服务器端涉及到的所有<code>Handler</code></p>*/publicclassServerHandlerInitializerextendsChannelInitializer<SocketChannel>{protectedvoidinitChannel(SocketChannelch)throwsException{ch.pipeline().addLast("idleStateHandler",newIdleStateHandler(5,0,0));ch.pipeline().addLast("idleStateTrigger",newServerIdleStateTrigger());ch.pipeline().addLast("frameDecoder",newLengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));ch.pipeline().addLast("frameEncoder",newLengthFieldPrepender(4));ch.pipeline().addLast("decoder",newStringDecoder());ch.pipeline().addLast("encoder",newStringEncoder());ch.pipeline().addLast("bizHandler",newServerBizHandler());}}
注：
new IdleStateHandler(5, 0, 0)该handler代表如果在5秒内没有收到来自客户端的任何数据包（包括但不限于心跳包），将会主动断开与该客户端的连接。
TcpServer —— 服务器端
publicclassTcpServer{privateintport;privateServerHandlerInitializerserverHandlerInitializer;publicTcpServer(intport){this.port=port;this.serverHandlerInitializer=newServerHandlerInitializer();}publicvoidstart(){EventLoopGroupbossGroup=newNioEventLoopGroup(1);EventLoopGroupworkerGroup=newNioEventLoopGroup();try{ServerBootstrapbootstrap=newServerBootstrap();bootstrap.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class).childHandler(this.serverHandlerInitializer);//绑定端口，开始接收进来的连接ChannelFuturefuture=bootstrap.bind(port).sync();System.out.println("Serverstartlistenat"+port);future.channel().closeFuture().sync();}catch(Exceptione){bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();e.printStackTrace();}}publicstaticvoidmain(String[]args)throwsException{intport=2222;newTcpServer(port).start();}}
至此，所有代码已经编写完毕。
测试
首先启动客户端，再启动服务器端。启动完成后，在客户端的控制台上，可以看到打印如下类似日志：
客户端控制台输出的日志
在服务器端可以看到控制台输出了类似如下的日志：
服务器端控制台输出的日志
可以看到，客户端在发送4个心跳包后，第5个包因为等待时间较长，等到真正发送的时候，发现连接已断开了；而服务器端收到客户端的4个心跳数据包后，迟迟等不到下一个数据包，所以果断断开该连接。
在测试过程中，有可能会出现如下情况：
异常情况
出现这种情况的原因是：在连接已断开的情况下，仍然向服务器端发送心跳包。虽然在发送心跳包之前会使用判断连接是否可用，但也有可能上一刻判断结果为可用，但下一刻发送数据包之前，连接就断了。
目前尚未找到优雅处理这种情况的方案，各位看官如果有好的解决方案，还望不吝赐教。拜谢！！！
断线重连
断线重连这里就不过多介绍，相信各位都知道是怎么回事。这里只说大致思路，然后直接上代码。
实现思路
客户端在监测到与服务器端的连接断开后，或者一开始就无法连接的情况下，使用指定的重连策略进行重连操作，直到重新建立连接或重试次数耗尽。
对于如何监测连接是否断开，则是通过重写ChannelInboundHandler#channelInactive来实现，但连接不可用，该方法会被触发，所以只需要在该方法做好重连工作即可。
代码实现
注：以下代码都是在上面心跳机制的基础上修改/添加的。
因为断线重连是客户端的工作，所以只需对客户端代码进行修改。
重试策略
RetryPolicy —— 重试策略接口
publicinterfaceRetryPolicy{/***Calledwhenanoperationhasfailedforsomereason.Thismethodshouldreturn*truetomakeanotherattempt.**@paramretryCountthenumberoftimesretriedsofar(0thefirsttime)*@returntrue/false*/booleanallowRetry(intretryCount);/***getsleeptimeinmsofcurrentretrycount.**@paramretryCountcurrentretrycount*@returnthetimetosleep*/longgetSleepTimeMs(intretryCount);}
ExponentialBackOffRetry —— 重连策略的默认实现
/***<p>Retrypolicythatretriesasetnumberoftimeswithincreasingsleeptimebetweenretries</p>*/publicclassExponentialBackOffRetryimplementsRetryPolicy{privatestaticfinalintMAX_RETRIES_LIMIT=29;privatestaticfinalintDEFAULT_MAX_SLEEP_MS=Integer.MAX_VALUE;privatefinalRandomrandom=newRandom();privatefinallongbaseSleepTimeMs;privatefinalintmaxRetries;privatefinalintmaxSleepMs;publicExponentialBackOffRetry(intbaseSleepTimeMs,intmaxRetries){this(baseSleepTimeMs,maxRetries,DEFAULT_MAX_SLEEP_MS);}publicExponentialBackOffRetry(intbaseSleepTimeMs,intmaxRetries,intmaxSleepMs){this.maxRetries=maxRetries;this.baseSleepTimeMs=baseSleepTimeMs;this.maxSleepMs=maxSleepMs;}@OverridepublicbooleanallowRetry(intretryCount){if(retryCount<maxRetries){returntrue;}returnfalse;}@OverridepubliclonggetSleepTimeMs(intretryCount){if(retryCount<0){thrownewIllegalArgumentException("retriescountmustgreaterthan0.");}if(retryCount>MAX_RETRIES_LIMIT){System.out.println(String.format("maxRetriestoolarge(%d).Pinningto%d",maxRetries,MAX_RETRIES_LIMIT));retryCount=MAX_RETRIES_LIMIT;}longsleepMs=baseSleepTimeMs*Math.max(1,random.nextInt(1<<retryCount));if(sleepMs>maxSleepMs){System.out.println(String.format("Sleepextensiontoolarge(%d).Pinningto%d",sleepMs,maxSleepMs));sleepMs=maxSleepMs;}returnsleepMs;}}
ReconnectHandler—— 重连处理器
@ChannelHandler.SharablepublicclassReconnectHandlerextendsChannelInboundHandlerAdapter{privateintretries=0;privateRetryPolicyretryPolicy;privateTcpClienttcpClient;publicReconnectHandler(TcpClienttcpClient){this.tcpClient=tcpClient;}@OverridepublicvoidchannelActive(ChannelHandlerContextctx)throwsException{System.out.println("Successfullyestablishedaconnectiontotheserver.");retries=0;ctx.fireChannelActive();}@OverridepublicvoidchannelInactive(ChannelHandlerContextctx)throwsException{if(retries==0){System.err.println("LosttheTCPconnectionwiththeserver.");ctx.close();}booleanallowRetry=getRetryPolicy().allowRetry(retries);if(allowRetry){longsleepTimeMs=getRetryPolicy().getSleepTimeMs(retries);System.out.println(String.format("Trytoreconnecttotheserverafter%dms.Retrycount:%d.",sleepTimeMs,++retries));finalEventLoopeventLoop=ctx.channel().eventLoop();eventLoop.schedule(()->{System.out.println("Reconnecting...");tcpClient.connect();},sleepTimeMs,TimeUnit.MILLISECONDS);}ctx.fireChannelInactive();}privateRetryPolicygetRetryPolicy(){if(this.retryPolicy==null){this.retryPolicy=tcpClient.getRetryPolicy();}returnthis.retryPolicy;}}
ClientHandlersInitializer
在之前的基础上，添加了重连处理器ReconnectHandler。
publicclassClientHandlersInitializerextendsChannelInitializer<SocketChannel>{privateReconnectHandlerreconnectHandler;privateEchoHandlerechoHandler;publicClientHandlersInitializer(TcpClienttcpClient){Assert.notNull(tcpClient,"TcpClientcannotbenull.");this.reconnectHandler=newReconnectHandler(tcpClient);this.echoHandler=newEchoHandler();}@OverrideprotectedvoidinitChannel(SocketChannelch)throwsException{ChannelPipelinepipeline=ch.pipeline();pipeline.addLast(this.reconnectHandler);pipeline.addLast(newLengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,4,0,4));pipeline.addLast(newLengthFieldPrepender(4));pipeline.addLast(newStringDecoder(CharsetUtil.UTF_8));pipeline.addLast(newStringEncoder(CharsetUtil.UTF_8));pipeline.addLast(newPinger());}}
TcpClient
在之前的基础上添加重连、重连策略的支持。
publicclassTcpClient{privateStringhost;privateintport;privateBootstrapbootstrap;/**重连策略*/privateRetryPolicyretryPolicy;/**将<code>Channel</code>保存起来,可用于在其他非handler的地方发送数据*/privateChannelchannel;publicTcpClient(Stringhost,intport){this(host,port,newExponentialBackOffRetry(1000,Integer.MAX_VALUE,60*1000));}publicTcpClient(Stringhost,intport,RetryPolicyretryPolicy){this.host=host;this.port=port;this.retryPolicy=retryPolicy;init();}/***向远程TCP服务器请求连接*/publicvoidconnect(){synchronized(bootstrap){ChannelFuturefuture=bootstrap.connect(host,port);future.addListener(getConnectionListener());this.channel=future.channel();}}publicRetryPolicygetRetryPolicy(){returnretryPolicy;}privatevoidinit(){EventLoopGroupgroup=newNioEventLoopGroup();//bootstrap可重用,只需在TcpClient实例化的时候初始化即可.bootstrap=newBootstrap();bootstrap.group(group).channel(NioSocketChannel.class).handler(newClientHandlersInitializer(TcpClient.this));}privateChannelFutureListenergetConnectionListener(){returnnewChannelFutureListener(){@OverridepublicvoidoperationComplete(ChannelFuturefuture)throwsException{if(!future.isSuccess()){future.channel().pipeline().fireChannelInactive();}}};}publicstaticvoidmain(String[]args){TcpClienttcpClient=newTcpClient("localhost",2222);tcpClient.connect();}}
测试
在测试之前，为了避开Connection reset by peer异常，可以稍微修改Pinger的ping()方法，添加if (second == 5)的条件判断。如下：
privatevoidping(Channelchannel){intsecond=Math.max(1,random.nextInt(baseRandom));if(second==5){second=6;}System.out.println("nextheartbeatwillsendafter"+second+"s.");ScheduledFuture<?>future=channel.eventLoop().schedule(newRunnable(){@Overridepublicvoidrun(){if(channel.isActive()){System.out.println("sendingheartbeattotheserver...");channel.writeAndFlush(ClientIdleStateTrigger.HEART_BEAT);}else{System.err.println("Theconnectionhadbroken,cancelthetaskthatwillsendaheartbeat.");channel.closeFuture();thrownewRuntimeException();}}},second,TimeUnit.SECONDS);future.addListener(newGenericFutureListener(){@OverridepublicvoidoperationComplete(Futurefuture)throwsException{if(future.isSuccess()){ping(channel);}}});}
启动客户端
先只启动客户端，观察控制台输出，可以看到类似如下日志：
断线重连测试——客户端控制台输出
可以看到，当客户端发现无法连接到服务器端，所以一直尝试重连。随着重试次数增加，重试时间间隔越大，但又不想无限增大下去，所以需要定一个阈值，比如60s。如上图所示，当下一次重试时间超过60s时，会打印Sleep extension too large(*). Pinning to 60000，单位为ms。出现这句话的意思是，计算出来的时间超过阈值（60s），所以把真正睡眠的时间重置为阈值（60s）。
启动服务器端
接着启动服务器端，然后继续观察客户端控制台输出。
断线重连测试——服务器端启动后客户端控制台输出
可以看到，在第9次重试失败后，第10次重试之前，启动的服务器，所以第10次重连的结果为，即成功连接到服务器。接下来因为还是不定时服务器，所以出现断线重连、断线重连的循环。
扩展
在不同环境，可能会有不同的重连需求。有不同的重连需求的，只需自己实现RetryPolicy接口，然后在创建TcpClient的时候覆盖默认的重连策略即可。
热门内容：
Java 项目权威排名：Nacos 未上版，Gradle 排名第二，Maven 排名 28
SpringBoot项目，如何优雅的把接口参数中的空白值替换为null值？
JDK 16 即将发布，新特性速览！
微服务架构中配置中心的选择
最近面试BAT，整理一份面试资料《Java面试BAT通关手册》，覆盖了Java核心技术、JVM、Java并发、SSM、微服务、数据库、数据结构等等。获取方式：点“在看”，关注公众号并回复666领取，更多内容陆续奉上。明天见(｡･ω･｡)ﾉ
如果觉得《Netty实现心跳机制与断线重连》对你有帮助，请点赞、收藏，并留下你的观点哦！
