Netty 如何实现心跳机制与断线重连?

2020-05-07 16:05:37来源:博客园 阅读 ()

新老客户大回馈,云服务器低至5折

Netty 如何实现心跳机制与断线重连?

作者:sprinkle_liz
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}事件(未在指定时间内向服务器发送数据),然后向<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.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 —— 服务器端处理器集合的初始化类

/**
 * <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.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);
 ? ? ? ? ? ? ? ?}
 ? ? ? ? ? ?}
 ? ? ? ?});
 ? ?}

启动客户端

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

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

可以看到,当客户端发现无法连接到服务器端,所以一直尝试重连。随着重试次数增加,重试时间间隔越大,但又不想无限增大下去,所以需要定一个阈值,比如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的时候覆盖默认的重连策略即可。

完!!!

推荐去我的博客阅读更多:

1.Java JVM、集合、多线程、新特性系列教程

2.Spring MVC、Spring Boot、Spring Cloud 系列教程

3.Maven、Git、Eclipse、Intellij IDEA 系列工具教程

4.Java、后端、架构、阿里巴巴等大厂最新面试题

觉得不错,别忘了点赞+转发哦!


原文链接:https://www.cnblogs.com/javastack/p/12843678.html
如有疑问请与原作者联系

标签:

版权申明:本站文章部分自网络,如有侵权,请联系:west999com@outlook.com
特别注意:本站所有转载文章言论不代表本站观点,本站所提供的摄影照片,插画,设计作品,如需使用,请与原作者联系,版权归原作者所有

上一篇:【JAVA SE基础篇】5.eclipse的使用和运行

下一篇:JAVA8十大新特性详解