多条告白如次剧本只需引入一次
心跳体制
何为心跳
所谓心跳, 即在 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端发送一个心跳包。
/** * <p> * 用来捕捉{@link IdleState#WRITER_IDLE}事变（未在指定功夫内向效劳器发送数据），而后向<code>Server</code>端发送一个心跳包。 * </p> */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 —— 心跳放射器
/** * <p>存户端贯穿到效劳器端后，会轮回实行一个工作：随机等候几秒，而后ping一下Server端，即发送一个心跳包。</p> */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 —— 断连触发器
/** * <p>在规则功夫内未收到存户端的任何数据包, 将积极割断该贯穿</p> */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 —— 效劳器端的交易处置器
/** * <p>收到来自存户端的数据包后, 径直在遏制台打字与印刷出来.</p> */@ChannelHandler.Sharablepublic 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 —— 效劳器端处置器汇合的初始化类
/** * <p>用来初始化效劳器端波及到的一切<code>Handler</code></p> */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(); }}至此，一切代码仍旧编写结束。
尝试
开始启用存户端，再启用效劳器端。启用实行后，在存户端的遏制台上，不妨看到打字与印刷如次一致日记：
存户端遏制台输入的日记
在效劳器端不妨看到遏制台输入了一致如次的日记：
效劳器端遏制台输入的日记
不妨看到，存户端在发送4个心跳包后，第5个包由于等候功夫较长，比及真实发送的功夫，创造贯穿已割断了；而效劳器端收到存户端的4个心跳数据包后，迟迟等不到下一个数据包，以是顽强割断该贯穿。
特殊情景
在尝试进程中，有大概会展示如次情景：
特殊情景
展示这种情景的因为是：在贯穿已割断的情景下，仍旧向效劳器端发送心跳包。固然在发送心跳包之前会运用channel.isActive()确定贯穿能否可用，但也有大概上一刻确定截止为可用，但下一刻发送数据包之前，贯穿就断了。
暂时尚未找到优美处置这种情景的计划，诸位看官即使有好的处置计划，还望不惜指教。拜谢！！！
断线重连
断线重连这边就然而多引见，断定诸位都领会是如何回事。这边只说大概思绪，而后径直上代码。
实行思绪
存户端在监测到与效劳器端的贯穿割断后，大概一发端就没辙贯穿的情景下，运用指定的重连战略举行重连操纵，直到从新创造贯穿或重试度数耗尽。
对于怎样监测贯穿能否割断，则是经过重写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 —— 重连战略的默许实行
/** * <p>Retry policy that retries a set number of times with increasing sleep time between retries</p> */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.Sharablepublic 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); } } }); }启用存户端
先只启用存户端，查看遏制台输入，不妨看到一致如次日记：
断线重连尝试——存户端遏制台输入
不妨看到，当存户端创造没辙贯穿到效劳器端，以是从来试验重连。跟着重试度数减少，重试功夫间隙越大，但又不想无穷增大下来，以是须要定一个阈值，比方60s。如上海图书馆所示，当下一次重试功夫胜过60s时，会打字与印刷Sleep extension too large(*). Pinning to 60000，单元为ms。展示这句话的道理是，计划出来的功夫胜过阈值（60s），以是把真实安置的功夫重置为阈值（60s）。
启用效劳器端
接着启用效劳器端，而后连接查看存户端遏制台输入。
断线重连尝试——效劳器端启用后存户端遏制台输入
不妨看到，在第9次重试波折后，第10次重试之前，启用的效劳器，以是第10次重连的截止为Successfully established a connection to the server.，即胜利贯穿到效劳器。接下来由于仍旧大概时ping效劳器，以是展示断线重连、断线重连的轮回。