Java Netty

前言

在实现TCP长连接功能中，客户端断线重连是一个很常见的问题，当使用netty实现断线重连时，是否考虑过如下几个问题：

  • 如何监听到客户端和服务端连接断开 ?
  • 如何实现断线后重新连接 ?
  • netty客户端线程给多大比较合理 ?

其实上面都在做断线重连时所遇到的问题，而 “netty客户端线程给多大比较合理?” 这个问题更是在做断线重连时因一个异常引发的思考。下面讲讲整个过程：
内容主要涉及在客户端，但是为了能够运行整个程序，所以这里先给出服务端及公共的依赖和实体类。

相关代码

服务端及common代码

maven依赖：

  1. <dependencies>
  2.     <!--只是用到了spring-boot的日志框架-->
  3.     <dependency>
  4.         <groupId>org.springframework.boot</groupId>
  5.         <artifactId>spring-boot-starter</artifactId>
  6.         <version>2.4.1</version>
  7.     </dependency>
  9.     <dependency>
  10.         <groupId>io.netty</groupId>
  11.         <artifactId>netty-all</artifactId>
  12.         <version>4.1.56.Final</version>
  13.     </dependency>
  15.     <dependency>
  16.         <groupId>org.jboss.marshalling</groupId>
  17.         <artifactId>jboss-marshalling-serial</artifactId>
  18.         <version>2.0.10.Final</version>
  19.     </dependency>
  20. </dependencies>

服务端业务处理代码

主要用于记录打印当前客户端连接数，当接收到客户端信息后返回“hello netty”字符串

  1. @ChannelHandler.Sharable
  2. public class SimpleServerHandler extends ChannelInboundHandlerAdapter {
  3.     private static final InternalLogger log = InternalLoggerFactory.getInstance(SimpleServerHandler.class);
  4.     public static final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
  6.     @Override
  7.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  8.         channels.add(ctx.channel());
  9.         log.info("客户端连接成功: client address :{}", ctx.channel().remoteAddress());
  10.         log.info("当前共有{}个客户端连接", channels.size());
  11.     }
  13.     @Override
  14.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  15.         log.info("server channelRead:{}", msg);
  16.         ctx.channel().writeAndFlush("hello netty");
  17.     }
  19.     @Override
  20.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  21.         log.info("channelInactive: client close");
  22.     }
  24.     @Override
  25.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  26.         if (cause instanceof java.io.IOException) {
  27.             log.warn("exceptionCaught: client close");
  28.         } else {
  29.             cause.printStackTrace();
  30.         }
  31.     }
  32. }

服务端心跳检查代码

当接收心跳”ping”信息后，返回客户端’’pong”信息。如果客户端在指定时间内没有发送任何信息则关闭客户端。

  1. public class ServerHeartbeatHandler extends ChannelInboundHandlerAdapter {
  2.     private static final InternalLogger log = InternalLoggerFactory.getInstance(ServerHeartbeatHandler.class);
  4.     @Override
  5.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  6.         log.info("server channelRead:{}", msg);
  7.         if (msg.equals("ping")) {
  8.             ctx.channel().writeAndFlush("pong");
  9.         } else {
  10.             //由下一个handler处理,示例中则为SimpleServerHandler
  11.             ctx.fireChannelRead(msg);
  12.         }
  13.     }
  15.     @Override
  16.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  17.         if (evt instanceof IdleStateEvent) {
  18.             //该事件需要配合 io.netty.handler.timeout.IdleStateHandler使用
  19.             IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
  20.             if (idleStateEvent.state() == IdleState.READER_IDLE) {
  21.                 //超过指定时间没有读事件,关闭连接
  22.                 log.info("超过心跳时间,关闭和服务端的连接:{}", ctx.channel().remoteAddress());
  23.                 //ctx.channel().close();
  24.             }
  25.         } else {
  26.             super.userEventTriggered(ctx, evt);
  27.         }
  28.     }
  29. }

编解码工具类

主要使用jboss-marshalling-serial编解码工具，可自行查询其优缺点，这里只是示例使用。

  1. public final class MarshallingCodeFactory {
  2.     /** 创建Jboss marshalling 解码器 */
  3.     public static MarshallingDecoder buildMarshallingDecoder() {
  4.         //参数serial表示创建的是Java序列化工厂对象,由jboss-marshalling-serial提供
  5.         MarshallerFactory factory = Marshalling.getProvidedMarshallerFactory("serial");
  6.         MarshallingConfiguration configuration = new MarshallingConfiguration();
  7.         configuration.setVersion(5);
  8.         DefaultUnmarshallerProvider provider = new DefaultUnmarshallerProvider(factory, configuration);
  9.         return new MarshallingDecoder(provider, 1024);
  10.     }
  12.     /** 创建Jboss marshalling 编码器 */
  13.     public static MarshallingEncoder buildMarshallingEncoder() {
  14.         MarshallerFactory factory = Marshalling.getProvidedMarshallerFactory("serial");
  15.         MarshallingConfiguration configuration = new MarshallingConfiguration();
  16.         configuration.setVersion(5);
  17.         DefaultMarshallerProvider provider = new DefaultMarshallerProvider(factory, configuration);
  18.         return new MarshallingEncoder(provider);
  19.     }
  20. }

公共实体类

  1. public class UserInfo implements Serializable {
  2.     private static final long serialVersionUID = 6271330872494117382L;
  4.     private String username;
  5.     private int age;
  7.     public UserInfo() {
  8.     }
  10.     public UserInfo(String username, int age) {
  11.         this.username = username;
  12.         this.age = age;
  13.     }
  14.    //省略getter/setter/toString
  15. }

下面开始客户端断线重连以及问题思考。

客户端实现

  • 刚开始启动时需要进行同步连接，指定连接次数内没用通过则抛出异常，进程退出。
  • 客户端启动后，开启定时任务，模拟客户端数据发送。

客户端业务处理handler，接收到数据后，通过日志打印。

  1. public class SimpleClientHandler extends ChannelInboundHandlerAdapter {
  2.     private static final InternalLogger log = InternalLoggerFactory.getInstance(SimpleClientHandler.class);
  3.     private NettyClient client;
  5.     public SimpleClientHandler(NettyClient client) {
  6.         this.client = client;
  7.     }
  9.     @Override
  10.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  11.         log.info("client receive:{}", msg);
  12.     }
  13. }

封装连接方法、断开连接方法、getChannel()返回io.netty.channel.Channel用于向服务端发送数据。boolean connect()是一个同步连接方法，如果连接成功返回true，连接失败返回false。

  1. public class NettyClient {
  2.     private static final InternalLogger log = InternalLoggerFactory.getInstance(NettyClient.class);
  4.     private EventLoopGroup workerGroup;
  5.     private Bootstrap bootstrap;
  6.     private volatile Channel clientChannel;
  8.     public NettyClient() {
  9.         this(-1);
  10.     }
  12.     public NettyClient(int threads) {
  13.         workerGroup = threads > 0 ? new NioEventLoopGroup(threads) : new NioEventLoopGroup();
  14.         bootstrap = new Bootstrap();
  15.         bootstrap.group(workerGroup)
  16.                 .channel(NioSocketChannel.class)
  17.                 .option(ChannelOption.TCP_NODELAY, true)
  18.                 .option(ChannelOption.SO_KEEPALIVE, false)
  19.                 .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30000)
  20.                 .handler(new ClientHandlerInitializer(this));
  21.     }
  23.     public boolean connect() {
  24.         log.info("尝试连接到服务端: 127.0.0.1:8088");
  25.         try {
  26.             ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8088);
  28.             boolean notTimeout = channelFuture.awaitUninterruptibly(30, TimeUnit.SECONDS);
  29.             clientChannel = channelFuture.channel();
  30.             if (notTimeout) {
  31.                 if (clientChannel != null && clientChannel.isActive()) {
  32.                     log.info("netty client started !!! {} connect to server", clientChannel.localAddress());
  33.                     return true;
  34.                 }
  35.                 Throwable cause = channelFuture.cause();
  36.                 if (cause != null) {
  37.                     exceptionHandler(cause);
  38.                 }
  39.             } else {
  40.                 log.warn("connect remote host[{}] timeout {}s", clientChannel.remoteAddress(), 30);
  41.             }
  42.         } catch (Exception e) {
  43.             exceptionHandler(e);
  44.         }
  45.         clientChannel.close();
  46.         return false;
  47.     }
  49.     private void exceptionHandler(Throwable cause) {
  50.         if (cause instanceof ConnectException) {
  51.             log.error("连接异常:{}", cause.getMessage());
  52.         } else if (cause instanceof ClosedChannelException) {
  53.             log.error("connect error:{}", "client has destroy");
  54.         } else {
  55.             log.error("connect error:", cause);
  56.         }
  57.     }
  59.     public void close() {
  60.         if (clientChannel != null) {
  61.             clientChannel.close();
  62.         }
  63.         if (workerGroup != null) {
  64.             workerGroup.shutdownGracefully();
  65.         }
  66.     }
  68.     public Channel getChannel() {
  69.         return clientChannel;
  70.     }
  72.     static class ClientHandlerInitializer extends ChannelInitializer<SocketChannel> {
  73.         private static final InternalLogger log = InternalLoggerFactory.getInstance(NettyClient.class);
  74.         private NettyClient client;
  76.         public ClientHandlerInitializer(NettyClient client) {
  77.             this.client = client;
  78.         }
  80.         @Override
  81.         protected void initChannel(SocketChannel ch) throws Exception {
  82.             ChannelPipeline pipeline = ch.pipeline();
  83.             pipeline.addLast(MarshallingCodeFactory.buildMarshallingDecoder());
  84.             pipeline.addLast(MarshallingCodeFactory.buildMarshallingEncoder());
  85.             //pipeline.addLast(new IdleStateHandler(25, 0, 10));
  86.             //pipeline.addLast(new ClientHeartbeatHandler());
  87.             pipeline.addLast(new SimpleClientHandler(client));
  88.         }
  89.     }
  90. }

客户端启动类

  1. public class NettyClientMain {
  2.     private static final InternalLogger log = InternalLoggerFactory.getInstance(NettyClientMain.class);
  3.     private static final ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
  5.     public static void main(String[] args) {
  6.         NettyClient nettyClient = new NettyClient();
  7.         boolean connect = false;
  8.         //刚启动时尝试连接10次,都无法建立连接则不在尝试
  9.         //如果想在刚启动后,一直尝试连接,需要放在线程中,异步执行,防止阻塞程序
  10.         for (int i = 0; i < 10; i++) {
  11.             connect = nettyClient.connect();
  12.             if (connect) {
  13.                 break;
  14.             }
  15.             //连接不成功,隔5s之后重新尝试连接
  16.             try {
  17.                 Thread.sleep(5000);
  18.             } catch (InterruptedException e) {
  19.                 e.printStackTrace();
  20.             }
  21.         }
  23.         if (connect) {
  24.             log.info("定时发送数据");
  25.             send(nettyClient);
  26.         } else {
  27.             nettyClient.close();
  28.             log.info("进程退出");
  29.         }
  30.     }
  32.     /** 定时发送数据 */
  33.     static void send(NettyClient client) {
  34.         scheduledExecutor.schedule(new SendTask(client,scheduledExecutor), 2, TimeUnit.SECONDS);
  35.     }
  36. }

客户端断线重连

断线重连需求：

  • 服务端和客户端之间网络异常，或响应超时（例如有个很长时间的fullGC），客户端需要主动重连其他节点。
  • 服务端宕机时或者和客户端之间发生任何异常时，客户端需要主动重连其他节点。
  • 服务端主动向客户端发送（服务端）下线通知时，客户端需要主动重连其他节点。

    如何监听到客户端和服务端连接断开 ?

    netty的io.netty.channel.ChannelInboundHandler接口中提供了许多重要的接口方法。为了避免实现全部的接口方法，可以通过继承io.netty.channel.ChannelInboundHandlerAdapter来重写相应的方法即可。
  1. void channelInactive(ChannelHandlerContext ctx);在客户端关闭时被调用，表示客户端断开连接。当如下几种情况发生时会触发：
  • 客户端在正常active状态下，主动调用channel或者ctx的close方法。
  • 服务端主动调用channel或者ctx的close方法关闭客户端的连接 。
  • 发生java.io.IOException（一般情况下是双方连接断开）或者java.lang.OutOfMemoryError（4.1.52版本中新增）时
  1. void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception;则是在入栈发生任何异常时被调用。如果异常是java.io.IOException或者java.lang.OutOfMemoryError(4.1.52版本新增)时，还会触发channelInactive方法，也就是上面channelInactive被触发的第3条情况。
  2. 心跳检查也是检查客户端与服务端之间连接状态的必要方式，因为在一些状态下，两端实际上已经断开连接，但客户端无法感知，这时候就需要通过心跳来判断两端的连接状态。心跳可以是客户端心跳和服务端心跳。
  • 客户端心跳：即为客户端发送心跳ping信息，服务端回复pong信息。这样在指定时间内，双方有数据交互则认为是正常连接状态。
  • 服务端心跳：则是服务端向客户端发送ping信息，客户端回复pong信息。在指定时间内没有收到回复，则认为对方下线。

netty提供了非常简单的心跳检查方式，只需要在channel的handler链上，添加io.netty.handler.timeout.IdleStateHandler即可实现。
IdleStateHandler有如下几个重要的参数：

  • readerIdleTimeSeconds，读超时. 即当在指定的时间间隔内没有从 Channel 读取到数据时，会触发一个READER_IDLE的IdleStateEvent 事件.
  • writerIdleTimeSeconds，写超时. 即当在指定的时间间隔内没有数据写入到 Channel 时，会触发一个WRITER_IDLE的IdleStateEvent事件.
  • allIdleTimeSeconds，读/写超时. 即当在指定的时间间隔内没有读或写操作时，会触发一个ALL_IDLE的IdleStateEvent 事件.

为了能够监听到这些事件的触发，还需要重写ChannelInboundHandler#userEventTriggered(ChannelHandlerContext ctx, Object evt)方法，通过参数evt判断事件类型。在指定的时间类如果没有读写则发送一条心跳的ping请求，在指定时间内没有收到读操作则任务已经和服务端断开连接。则调用channel或者ctx的close方法，使客户端Handler执行channelInactive方法。
到这里看来只要在channelInactive和exceptionCaught两个方法中实现自己的重连逻辑即可，但是遇到了第一个坑，重连方法执行了两次。
先看示例代码和结果，在com.bruce.netty.rpc.client.SimpleClientHandler中添加如下代码：

  1. public class SimpleClientHandler extends ChannelInboundHandlerAdapter {
  2.     private static final InternalLogger log = InternalLoggerFactory.getInstance(SimpleClientHandler.class);
  3.     //省略部分代码......
  4.     /** 客户端正常下线时执行该方法 */
  5.     @Override
  6.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  7.         log.warn("channelInactive:{}", ctx.channel().localAddress());
  8.         reconnection(ctx);
  9.     }
  11.     /** 入栈发生异常时执行exceptionCaught */
  12.     @Override
  13.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  14.         if (cause instanceof IOException) {
  15.             log.warn("exceptionCaught:客户端[{}]和远程断开连接", ctx.channel().localAddress());
  16.         } else {
  17.             log.error(cause);
  18.         }
  19.         reconnection(ctx);
  20.     }
  22.     private void reconnection(ChannelHandlerContext ctx) {
  23.         log.info("5s之后重新建立连接");
  24.         //暂时为空实现
  25.     }
  26. }

ClientHandlerInitializer 中添加io.netty.handler.timeout.IdleStateHandler用于心跳检查，ClientHeartbeatHandler用于监听心跳事件，接收心跳pong回复。

  1. static class ClientHandlerInitializer extends ChannelInitializer<SocketChannel> {
  2.     private static final InternalLogger log = InternalLoggerFactory.getInstance(NettyClient.class);
  3.     private NettyClient client;
  5.     public ClientHandlerInitializer(NettyClient client) {
  6.         this.client = client;
  7.     }
  9.     @Override
  10.     protected void initChannel(SocketChannel ch) throws Exception {
  11.         ChannelPipeline pipeline = ch.pipeline();
  12.         pipeline.addLast(MarshallingCodeFactory.buildMarshallingDecoder());
  13.         pipeline.addLast(MarshallingCodeFactory.buildMarshallingEncoder());
  14.         //25s内没有read操作则触发READER_IDLE事件
  15.         //10s内既没有read又没有write操作则触发ALL_IDLE事件
  16.         pipeline.addLast(new IdleStateHandler(25, 0, 10));
  17.         pipeline.addLast(new ClientHeartbeatHandler());
  18.         pipeline.addLast(new SimpleClientHandler(client));
  19.     }
  20. }

com.bruce.netty.rpc.client.ClientHeartbeatHandler

  1. public class ClientHeartbeatHandler extends ChannelInboundHandlerAdapter {
  2.     private static final InternalLogger log = InternalLoggerFactory.getInstance(ClientHeartbeatHandler.class);
  4.     @Override
  5.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  6.         if (msg.equals("pong")) {
  7.             log.info("收到心跳回复");
  8.         } else {
  9.             super.channelRead(ctx, msg);
  10.         }
  11.     }
  13.     @Override
  14.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  15.         if (evt instanceof IdleStateEvent) {
  16.             //该事件需要配合 io.netty.handler.timeout.IdleStateHandler使用
  17.             IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
  18.             if (idleStateEvent.state() == IdleState.ALL_IDLE) {
  19.                 //向服务端发送心跳检测
  20.                 ctx.writeAndFlush("ping");
  21.                 log.info("发送心跳数据");
  22.             } else if (idleStateEvent.state() == IdleState.READER_IDLE) {
  23.                 //超过指定时间没有读事件,关闭连接
  24.                 log.info("超过心跳时间,关闭和服务端的连接:{}", ctx.channel().remoteAddress());
  25.                 ctx.channel().close();
  26.             }
  27.         } else {
  28.             super.userEventTriggered(ctx, evt);
  29.         }
  30.     }
  31. }

先启动server端，再启动client端，待连接成功之后kill掉 server端进程。
2021-12-24-14-56-04-089145.png
通过客户端日志可以看出，先是执行了exceptionCaught方法然后执行了channelInactive方法，但是这两个方法中都调用了reconnection方法，导致同时执行了两次重连。
为什么执行了exceptionCaught方法又执行了channelInactive方法呢？
可以在exceptionCaught和channelInactive方法添加断点一步步查看源码
2021-12-24-14-56-04-486685.png
当NioEventLoop执行select操作之后，处理相应的SelectionKey，发生异常后，会调用AbstractNioByteChannel.NioByteUnsafe#handleReadException方法进行处理，并触发pipeline.fireExceptionCaught(cause)，最终调用到用户handler的fireExceptionCaught方法。

  1. private void handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf, Throwable cause, boolean close,
  2.                                  RecvByteBufAllocator.Handle allocHandle) {
  3.   if (byteBuf != null) {
  4.     if (byteBuf.isReadable()) {
  5.       readPending = false;
  6.       pipeline.fireChannelRead(byteBuf);
  7.     } else {
  8.       byteBuf.release();
  9.     }
  10.   }
  11.   allocHandle.readComplete();
  12.   pipeline.fireChannelReadComplete();
  13.   pipeline.fireExceptionCaught(cause);
  15.   // If oom will close the read event, release connection.
  16.   // See https://github.com/netty/netty/issues/10434
  17.   if (close || cause instanceof OutOfMemoryError || cause instanceof IOException) {
  18.     closeOnRead(pipeline);
  19.   }
  20. }

该方法最后会判断异常类型，执行close连接的方法。在连接断线的场景中，这里即为java.io.IOException，所以执行了close方法，当debug到AbstractChannel.AbstractUnsafe#close(ChannelPromise, Throwable, ClosedChannelException, notify)方法中会发现最后又调用了AbstractUnsafe#fireChannelInactiveAndDeregister方法，继续debug最后则会执行自定义的fireChannelInactive方法。
到这里可以总结一个知识点：netty中当执行到handler地fireExceptionCaught方法时，可能会继续触发到fireChannelInactive，也可能不会触发fireChannelInactive。
除了netty根据异常类型判断是否执行close方法外，其实开发人员也可以自己通过ctx或者channel去调用close方法，代码如下：

  1. @Override
  2. public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  3.     if (cause instanceof IOException) {
  4.         log.warn("exceptionCaught:客户端[{}]和远程断开连接", ctx.channel().localAddress());
  5.     } else {
  6.         log.error(cause);
  7.     }
  8.     //ctx.close();
  9.     ctx.channel().close();
  10. }

但这种显示调用close方法，是否一定会触发调用fireChannelInactive呢？
如果是，那么只需要在exceptionCaught中调用close方法，fireChannelInactive中做重连的逻辑即可！！
通过日志观察到，在exceptionCaught中调用close方法每次都会调用fireChannelInactive方法。但是查看源码，可以看到这是不一定的，因为在AbstractChannel.AbstractUnsafe#close(ChannelPromise,Throwable, ClosedChannelException, notify)中会调用io.netty.channel.Channel#isActive进行判断，只有为true，才会执行fireChannelInactive方法。

  1. //io.netty.channel.socket.nio.NioSocketChannel#isActive
  2. @Override
  3. public boolean isActive() {
  4.     SocketChannel ch = javaChannel();
  5.     return ch.isOpen() && ch.isConnected();
  6. }

如何解决同时执行两次问题呢？

在netty初始化时，都会添加一系列的handler处理器，这些handler实际上会在netty创建Channel对象(NioSocketChannel)时，被封装在DefaultChannelPipeline中，而DefaultChannelPipeline实际上是一个双向链表，头节点为TailContext，尾节点为TailContext，而中间的节点则是添加的一个个handler（被封装成DefaultChannelHandlerContext），当执行Pipeline上的方法时，会从链表上遍历handler执行，因此当执行exceptionCaught方法时，只需要提前移除链表上自定义的Handler则无法执行fireChannelInactive方法。
2021-12-24-14-56-04-870366.png
最后实现代码如下：

  1. public class SimpleClientHandler extends ChannelInboundHandlerAdapter {
  3.     private static final InternalLogger log = InternalLoggerFactory.getInstance(SimpleClientHandler.class);
  5.     @Override
  6.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  7.         log.warn("channelInactive:{}", ctx.channel().localAddress());
  8.         ctx.pipeline().remove(this);
  9.         ctx.channel().close();
  10.         reconnection(ctx);
  11.     }
  13.     @Override
  14.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  15.         if (cause instanceof IOException) {
  16.             log.warn("exceptionCaught:客户端[{}]和远程断开连接", ctx.channel().localAddress());
  17.         } else {
  18.             log.error(cause);
  19.         }
  20.         ctx.pipeline().remove(this);
  21.         //ctx.close();
  22.         ctx.channel().close();
  23.         reconnection(ctx);
  24.     }
  25. }

执行效果如下，可以看到当发生异常时，只是执行了exceptionCaught方法，并且通过channel关闭了上一次连接资源，也没有执行当前handler的fireChannelInactive方法。
2021-12-24-14-56-05-381521.png

如何实现断线后重新连接 ?

通过上面分析，已经知道在什么方法中实现自己的重连逻辑，但是具体该怎么实现呢，怀着好奇的心态搜索了一下各大码友的实现方案。大多做法是通过ctx.channel().eventLoop().schedule添加一个定时任务调用客户端的连接方法。实现代码如下：

  1. private void reconnection(ChannelHandlerContext ctx) {
  2.     log.info("5s之后重新建立连接");
  3.     ctx.channel().eventLoop().schedule(new Runnable() {
  4.         @Override
  5.         public void run() {
  6.             boolean connect = client.connect();
  7.             if (connect) {
  8.                 log.info("重新连接成功");
  9.             } else {
  10.                 reconnection(ctx);
  11.             }
  12.         }
  13.     }, 5, TimeUnit.SECONDS);
  14. }

测试：先启动server端，再启动client端，待连接成功之后kill掉 server端进程。客户端如期定时执行重连，但也就去茶水间倒杯水的时间，回来后发现了如下异常。

  1. ......省略14条相同的重试日志
  2. [2021-01-17 18:46:45.032] INFO   [nioEventLoopGroup-2-1] [com.bruce.netty.rpc.client.SimpleClientHandler] : 5s之后重新建立连接
  3. [2021-01-17 18:46:48.032] INFO   [nioEventLoopGroup-2-1] [com.bruce.netty.rpc.client.NettyClient] : 尝试连接到服务端: 127.0.0.1:8088
  4. [2021-01-17 18:46:50.038] ERROR   [nioEventLoopGroup-2-1] [com.bruce.netty.rpc.client.NettyClient] : 连接异常:Connection refused: no further information: /127.0.0.1:8088
  5. [2021-01-17 18:46:50.038] INFO   [nioEventLoopGroup-2-1] [com.bruce.netty.rpc.client.SimpleClientHandler] : 5s之后重新建立连接
  6. [2021-01-17 18:46:53.040] INFO   [nioEventLoopGroup-2-1] [com.bruce.netty.rpc.client.NettyClient] : 尝试连接到服务端: 127.0.0.1:8088
  7. [2021-01-17 18:46:53.048] ERROR   [nioEventLoopGroup-2-1] [com.bruce.netty.rpc.client.NettyClient] : connect error:
  8. io.netty.util.concurrent.BlockingOperationException: DefaultChannelPromise@10122121(incomplete)
  9.      at io.netty.util.concurrent.DefaultPromise.checkDeadLock(DefaultPromise.java:462)
  10.      at io.netty.channel.DefaultChannelPromise.checkDeadLock(DefaultChannelPromise.java:159)
  11.      at io.netty.util.concurrent.DefaultPromise.await0(DefaultPromise.java:667)
  12.      at io.netty.util.concurrent.DefaultPromise.awaitUninterruptibly(DefaultPromise.java:305)
  13.      at com.bruce.netty.rpc.client.NettyClient.connect(NettyClient.java:49)
  14.      at com.bruce.netty.rpc.client.SimpleClientHandler$1.run(SimpleClientHandler.java:65)
  15.      at io.netty.util.concurrent.PromiseTask.runTask(PromiseTask.java:98)
  16.      at io.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:170)
  17.      at io.netty.util.concurrent.AbstractEventExecutor.safeExecute$$$capture(AbstractEventExecutor.java:164)
  18.      at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java)
  19.      at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:472)
  20.      at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:500)

根据异常栈，可以发现是com.bruce.netty.rpc.client.NettyClient#connect方法中调用了等待方法

  1. boolean notTimeout = channelFuture.awaitUninterruptibly(20, TimeUnit.SECONDS);

而该方法内部会进行检测，是否在io线程上执行了同步等待，这会导致抛出异常BlockingOperationException。

  1. @Override
  2. protected void checkDeadLock() {
  3.     if (channel().isRegistered()) {
  4.         super.checkDeadLock();
  5.     }
  6. }
  8. protected void checkDeadLock() {
  9.     EventExecutor e = executor();
  10.     if (e != null && e.inEventLoop()) {
  11.         throw new BlockingOperationException(toString());
  12.     }
  13. }

奇怪的是为什么不是每次尝试重连都抛出该异常，而是每隔16次抛出一次呢?
联想到笔记本是8核处理器，而netty默认线程池是2 * c，就是16条线程，这之间似乎有些关联。
实际上在调用ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8088);，netty首先会创建一个io.netty.channel.Channel(示例中是NioSocketChannel)，然后通过io.netty.util.concurrent.EventExecutorChooserFactory.EventExecutorChooser依次选择一个NioEventLoop，将Channel绑定到NioEventLoop上。
2021-12-24-14-56-05-743905.png
io.netty.util.concurrent.SingleThreadEventExecutor#inEventLoop

  1. //Return true if the given Thread is executed in the event loop, false otherwise.
  2. @Override
  3. public boolean inEventLoop(Thread thread) {
  4.     return thread == this.thread;
  5. }

重连的方法是在一个NioEventLoop(也就是io线程)上被调用，第1次重连实际上是选择了第2个NioEventLoop，第2次重连实际上是选择了第3个NioEventLoop，以此类推，当一轮选择过后，重新选到第一个NioEventLoop时，boolean inEventLoop()返回true，则抛出了BlockingOperationException。

方案1

不要在netty的io线程上执行同步连接，使用单独的线程池定时执行重试，该线程还可以执行自己重连的业务逻辑操作，不阻塞io线程。（如果不需要业务操作之后销毁线程池）。
com.bruce.netty.rpc.client.SimpleClientHandler 修改reconnection方法

  1. private static ScheduledExecutorService SCHEDULED_EXECUTOR;
  3. private void initScheduledExecutor() {
  4.     if (SCHEDULED_EXECUTOR == null) {
  5.         synchronized (SimpleClientHandler.class) {
  6.             if (SCHEDULED_EXECUTOR == null) {
  7.                 SCHEDULED_EXECUTOR = Executors.newSingleThreadScheduledExecutor(r -> {
  8.                     Thread t = new Thread(r, "Client-Reconnect-1");
  9.                     t.setDaemon(true);
  10.                     return t;
  11.                 });
  12.             }
  13.         }
  14.     }
  15. }
  17. private void reconnection(ChannelHandlerContext ctx) {
  18.     log.info("5s之后重新建立连接");
  19.     initScheduledExecutor();
  21.     SCHEDULED_EXECUTOR.schedule(() -> {
  22.         boolean connect = client.connect();
  23.         if (connect) {
  24.             //连接成功,关闭线程池
  25.             SCHEDULED_EXECUTOR.shutdown();
  26.             log.info("重新连接成功");
  27.         } else {
  28.             reconnection(ctx);
  29.         }
  30.     }, 3, TimeUnit.SECONDS);
  31. }

方案2

可以在io线程上使用异步重连:
com.bruce.netty.rpc.client.NettyClient添加方法connectAsync方法，两者的区别在于connectAsync方法中没有调用channelFuture的同步等待方法。而是改成监听器(ChannelFutureListener)的方式，实际上这个监听器是运行在io线程上。

  1. public void connectAsync() {
  2.     log.info("尝试连接到服务端: 127.0.0.1:8088");
  3.     ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8088);
  4.     channelFuture.addListener((ChannelFutureListener) future -> {
  5.         Throwable cause = future.cause();
  6.         if (cause != null) {
  7.             exceptionHandler(cause);
  8.             log.info("等待下一次重连");
  9.             channelFuture.channel().eventLoop().schedule(this::connectAsync, 5, TimeUnit.SECONDS);
  10.         } else {
  11.             clientChannel = channelFuture.channel();
  12.             if (clientChannel != null && clientChannel.isActive()) {
  13.                 log.info("Netty client started !!! {} connect to server", clientChannel.localAddress());
  14.             }
  15.         }
  16.     });
  17. }

com.bruce.netty.rpc.client.SimpleClientHandler

  1. public class SimpleClientHandler extends ChannelInboundHandlerAdapter {
  2.     private static final InternalLogger log = InternalLoggerFactory.getInstance(SimpleClientHandler.class);
  3.     private NettyClient client;
  5.     public SimpleClientHandler(NettyClient client) {
  6.         this.client = client;
  7.     }
  9.     @Override
  10.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  11.         log.info("client receive:{}", msg);
  12.     }
  14.     @Override
  15.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  16.         log.warn("channelInactive:{}", ctx.channel().localAddress());
  17.         ctx.pipeline().remove(this);
  18.         ctx.channel().close();
  19.         reconnectionAsync(ctx);
  20.     }
  22.     @Override
  23.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  24.         if (cause instanceof IOException) {
  25.             log.warn("exceptionCaught:客户端[{}]和远程断开连接", ctx.channel().localAddress());
  26.         } else {
  27.             log.error(cause);
  28.         }
  29.         ctx.pipeline().remove(this);
  30.         ctx.close();
  31.         reconnectionAsync(ctx);
  32.     }
  34.     private void reconnectionAsync(ChannelHandlerContext ctx) {
  35.         log.info("5s之后重新建立连接");
  36.         ctx.channel().eventLoop().schedule(new Runnable() {
  37.             @Override
  38.             public void run() {
  39.                 client.connectAsync();
  40.             }
  41.         }, 5, TimeUnit.SECONDS);
  42.     }
  43. }

netty客户端线程给多大比较合理 ?

netty中一个NioEventLoopGroup默认创建的线程数是cpu核心数 * 2 ，这些线程都是用于io操作，那么对于客户端应用程序来说真的需要这么多io线程么？
通过上面分析BlockingOperationException异常时分析到，实际上netty在创建一个Channel对象后只会从NioEventLoopGroup中选择一个NioEventLoop来绑定，只有创建多个Channel才会依次选择下一个NioEventLoop，也就是说一个Channel只会对应一个NioEventLoop，而NioEventLoop可以绑定多个Channel。

  1. 对于客户端来说，如果只是连接的一个server节点，那么只要设置1条线程即可。即使出现了断线重连，在连接断开之后，之前的Channel会从NioEventLoop移除。重连之后，仍然只会在仅有的一个NioEventLoop注册一个新的Channel。

  2. 如果客户端同时如下方式多次调用io.netty.bootstrap.Bootstrap#connect(String inetHost, int inetPort)连接多个Server节点，那么线程可以设置大一点，但不要超过2*c，而且只要出现断线重连，同样不能保证每个NioEventLoop都会绑定一个客户端Channel。

    1. public boolean connect() {
    2.  try {
    3.      ChannelFuture channelFuture1 = bootstrap.connect("127.0.0.1", 8088);
    4.      ChannelFuture channelFuture2 = bootstrap.connect("127.0.0.1", 8088);
    5.      ChannelFuture channelFuture3 = bootstrap.connect("127.0.0.1", 8088);
    6.  } catch (Exception e) {
    7.      exceptionHandler(e);
    8.  }
    9.  clientChannel.close();
    10.  return false;
    11. }

  3. 如果netty客户端线程数设置大于1有什么影响么？

明显的异常肯定是不会有的，但是造成资源浪费，首先会创建多个NioEventLoop对象，但NioEventLoop是处于非运行状态。一旦出现断线重连，那么重新连接时，下一个NioEventLoop则会被选中，并创建/启动线程一直处于runnable状态。而上一个NioEventLoop也是一直处于runnable状态，由于上一个Channel已经被close，所以会造成每次select结果都是空的，没有意义的空轮询。
如下则是netty客户端使用默认线程数，4次断线重连后一共创建的5条NioEventLoop线程，但是实际上只有第5条线程在执行读写操作。
2021-12-24-14-56-06-007260.png2021-12-24-14-56-06-437679.png

  1. 如果客户端存在耗时的业务逻辑，应该单独使用业务线程池，避免在netty的io线程中执行耗时逻辑处理。