把自己的网站卖给别人后对方做违法,分销商城功能,乐清网站网络公司,沈阳网站seo外包来源#xff1a;www.jianshu.com/p/1a28e48edd92心跳机制何为心跳所谓心跳, 即在 TCP 长连接中, 客户端和服务器之间定期发送的一种特殊的数据包, 通知对方自己还在线, 以确保 TCP 连接的有效性.注#xff1a;心跳包还有另一个作用#xff0c;经常被忽略#xff0c;即… 来源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端发送一个心跳包。/*** p* 用于捕获{link IdleState#WRITER_IDLE}事件未在指定时间内向服务器发送数据然后向codeServer/code端发送一个心跳包。* /p*/
public class ClientIdleStateTrigger extends ChannelInboundHandlerAdapter {public static final String HEART_BEAT heart beat!;Overridepublic 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 serverctx.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;Overridepublic 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() {Overridepublic 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() {Overridepublic void operationComplete(Future future) throws Exception {if (future.isSuccess()) {ping(channel);}}});}Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {// 当Channel已经断开的情况下, 仍然发送数据, 会抛异常, 该方法会被调用.cause.printStackTrace();ctx.close();}
}
ClientHandlersInitializer —— 客户端处理器集合的初始化类public class ClientHandlersInitializer extends ChannelInitializerSocketChannel {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();}Overrideprotected 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;/** 将codeChannel/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 {Overridepublic 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.Sharable
public class ServerBizHandler extends SimpleChannelInboundHandlerString {private final String REC_HEART_BEAT I had received the heart beat!;Overrideprotected 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();}}Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println(Established connection with the remote client.);// do somethingctx.fireChannelActive();}Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {System.out.println(Disconnected with the remote client.);// do somethingctx.fireChannelInactive();}Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {cause.printStackTrace();ctx.close();}
}
ServerHandlerInitializer —— 服务器端处理器集合的初始化类/*** p用于初始化服务器端涉及到的所有codeHandler/code/p*/
public class ServerHandlerInitializer extends ChannelInitializerSocketChannel {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个心跳数据包后迟迟等不到下一个数据包所以果断断开该连接。在测试过程中有可能会出现如下情况异常情况出现这种情况的原因是在连接已断开的情况下仍然向服务器端发送心跳包。虽然在发送心跳包之前会使用判断连接是否可用但也有可能上一刻判断结果为可用但下一刻发送数据包之前连接就断了。目前尚未找到优雅处理这种情况的方案各位看官如果有好的解决方案还望不吝赐教。拜谢断线重连断线重连这里就不过多介绍相信各位都知道是怎么回事。这里只说大致思路然后直接上代码。实现思路客户端在监测到与服务器端的连接断开后或者一开始就无法连接的情况下使用指定的重连策略进行重连操作直到重新建立连接或重试次数耗尽。对于如何监测连接是否断开则是通过重写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 —— 重连策略的默认实现/*** pRetry 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;}Overridepublic boolean allowRetry(int retryCount) {if (retryCount maxRetries) {return true;}return false;}Overridepublic 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;}Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.out.println(Successfully established a connection to the server.);retries 0;ctx.fireChannelActive();}Overridepublic 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 ChannelInitializerSocketChannel {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();}Overrideprotected 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;/** 将codeChannel/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() {Overridepublic 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() {Overridepublic 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() {Overridepublic 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次重连的结果为即成功连接到服务器。接下来因为还是不定时服务器所以出现断线重连、断线重连的循环。扩展在不同环境可能会有不同的重连需求。有不同的重连需求的只需自己实现RetryPolicy接口然后在创建TcpClient的时候覆盖默认的重连策略即可。
