龙空技术网

龙空技术网

面试官问：服务的心跳机制与断线重连，Netty底层是怎么实现的？懵了

Meta多元宇宙 877

前言：

此刻同学们对“netty连接池断线重连”大体比较注重，我们都想要知道一些“netty连接池断线重连”的相关资讯。那么小编同时在网上汇集了一些对于“netty连接池断线重连””的相关文章，希望姐妹们能喜欢，各位老铁们快快来学习一下吧！

★★★建议星标我们★★★



2020年Java原创面试题库连载中

【000期】Java最全面试题库思维导图

【020期】JavaSE系列面试题汇总（共18篇）

【028期】JavaWeb系列面试题汇总（共10篇）

【042期】JavaEE系列面试题汇总（共13篇）

【049期】数据库系列面试题汇总（共6篇）

【053期】中间件系列面试题汇总（共3篇）

【065期】数据结构与算法面试题汇总（共11篇）

【076期】分布式面试题汇总（共10篇）

【077期】综合面试题系列（一）

【078期】综合面试题系列（二）

【079期】综合面试题系列（三）

【080期】综合面试题系列（四）

【081期】综合面试题系列（五）

【082期】综合面试题系列（六）

【083期】综合面试题系列（七）

【084期】综合面试题系列（八）

【085期】综合面试题系列（九）

【086期】综合面试题系列（十）

【087期】综合面试题系列（十一）

【088期】综合面试题系列（十二）

【089期】综合面试题系列（十三）

更多内容，点击上面蓝字查看


心跳机制何为心跳

所谓心跳, 即在 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.not(tcpClient, "TcpClient can not be .");
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();
 }
}

至此，所有代码已经编写完毕。

测试

首先启动客户端，再启动服务器端。启动完成后，在客户端的控制台上，可以看到打印如下类似日志：

客户端控制台输出的日志

在服务器端可以看到控制台输出了类似如下的日志：

img

服务器端控制台输出的日志

可以看到，客户端在发送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 —— 重连策略的默认实现
/**
 * <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 == ) {
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.not(tcpClient, "TcpClient can not be .");
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）。

启动服务器端

接着启动服务器端，然后继续观察客户端控制台输出。

img

断线重连测试——服务器端启动后客户端控制台输出

可以看到，在第9次重试失败后，第10次重试之前，启动的服务器，所以第10次重连的结果为，即成功连接到服务器。接下来因为还是不定时服务器，所以出现断线重连、断线重连的循环。

扩展

在不同环境，可能会有不同的重连需求。有不同的重连需求的，只需自己实现RetryPolicy接口，然后在创建TcpClient的时候覆盖默认的重连策略即可。

来源 | 


之前，给大家发过三份Java面试宝典，这次新增了一份，目前总共是四份面试宝典，相信在跳槽前一个月按照面试宝典准备准备，基本没大问题。
《java面试宝典5.0》(初中级)
《350道Java面试题：整理自100+公司》（中高级）
《资深java面试宝典-视频版》（资深）
《Java[BAT]面试必备》（资深）
分别适用于初中级，中高级，资深级工程师的面试复习。
内容包含java基础、javaweb、mysql性能优化、JVM、锁、百万并发、消息队列，高性能缓存、反射、Spring全家桶原理、微服务、Zookeeper、数据结构、限流熔断降级等等。
看到这里，证明有所收获

版权声明： 本站文章均来自互联网搜集，如有侵犯您的权益，请联系我们删除，谢谢。

标签： #netty连接池断线重连

上一篇手把手教会将 Windows 窗体应用从.NET 迁移到 .NET SDK/.NET 6 格式

下一篇做通信的你，这九大5G问题你不懂就OUT了

相关文章