Netty客户端自动重连线程阻塞问题及优化方案咨询

问题背景 #

我有一个已稳定运行多年的Netty客户端，实现了基础连接及自动重连功能。近期出现异常：连接进入close_wait状态，线程永久阻塞，提前设置的3秒连接超时完全未生效。

现有代码实现 #

Bootstrap初始化 #

Bootstrap bootstrap = new Bootstrap();

bootstrap.group(new NioEventLoopGroup(NUM_OF_WORKER_THREADS, new NamedThreadFactory(client.hostname+"-%d")))
    .channel(NioSocketChannel.class)
    .handler(new MyChannelInitializer(sslContext, client))
    .option(ChannelOption.SO_KEEPALIVE, true)
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);

连接逻辑 #

public void connect() { 
    try {
        ChannelFuture cf = bootstrap.connect(hostname, port).sync().await();
    } catch (Exception e) {
        if (group != null && !group.isShutdown()) {
            logger.error("Shutting down Event loop group: {}, host: {}" , group, hostname);
            group.shutdownGracefully();
        }
        throw new Exception("Connection failed to " + hostname, e);
    }
}

原重连逻辑 #

public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
    super.channelUnregistered(ctx);
    ctx.channel().close();
    //Connect again
    connect();
}

异常现象与线程栈 #

线程永久阻塞，对应的线程dump如下：

"server-callback-worker-1" #214 prio=5 os_prio=0 cpu=1322.64ms elapsed=83237.37s tid=0x00007f1b0c005800 nid=0x1a9e89 in Object.wait()  [0x00007f1b5c1fd000]
   java.lang.Thread.State: WAITING (on object monitor)
    at java.lang.Object.wait(java.base@11.0.22/Native Method)
    - waiting on <no object reference available>
    at java.lang.Object.wait(java.base@11.0.22/Object.java:328)
    at io.netty.util.concurrent.DefaultPromise.await(DefaultPromise.java:254)
    - waiting to re-lock in wait() <0x0000000461b0aba8> (a io.netty.bootstrap.AbstractBootstrap$PendingRegistrationPromise)
    at io.netty.channel.DefaultChannelPromise.await(DefaultChannelPromise.java:131)
    at io.netty.channel.DefaultChannelPromise.await(DefaultChannelPromise.java:30)
    at io.netty.util.concurrent.DefaultPromise.sync(DefaultPromise.java:405)
    at io.netty.channel.DefaultChannelPromise.sync(DefaultChannelPromise.java:119)
    at io.netty.channel.DefaultChannelPromise.sync(DefaultChannelPromise.java:30)
    at com.example.Client.connect(Client.java:113)
    at com.example.Client.reconnect(Client.java:141)
    at com.example.handler.ClientHandler.channelUnregistered(ClientHandler.java:72)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelUnregistered(AbstractChannelHandlerContext.java:219)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelUnregistered(AbstractChannelHandlerContext.java:195)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelUnregistered(AbstractChannelHandlerContext.java:188)
    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelUnregistered(DefaultChannelPipeline.java:1388)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelUnregistered(AbstractChannelHandlerContext.java:215)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelUnregistered(AbstractChannelHandlerContext.java:195)
    at io.netty.channel.DefaultChannelPipeline.fireChannelUnregistered(DefaultChannelPipeline.java:821)
    at io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:821)
    at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
    at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
    at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:566)
    at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
    at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
    at java.lang.Thread.run(java.base@11.0.22/Thread.java:829)

问题排查与解决方案 #

根因分析 #

确实存在线程阻塞导致的超时机制失效：

• Netty的channelUnregistered回调运行在EventLoop线程中，而connect()方法里调用的sync().await()会让该EventLoop线程进入阻塞等待状态。
• Netty的连接超时检测任务同样由该EventLoop线程负责执行，一旦线程被阻塞在await()上，超时任务无法被调度执行，最终导致3秒超时设置完全失效，线程永久等待。

解决方案评估 #

1. 专用Executor重构方案的正确性
   你尝试的使用scheduledExecutorService异步调度重连的方案是正确的：

   • 把重连逻辑从EventLoop线程转移到独立线程池执行，避免阻塞EventLoop，让EventLoop可以正常处理IO事件和超时检测任务。
   • 延迟3秒重连还能避免短时间内频繁重试导致的资源浪费。

2. 是否需要用Channel EventLoop调度？
   也可以使用EventLoop调度，但必须修改重连逻辑，不能再调用阻塞方法：

   • 改用ctx.executor().schedule(...)来调度重连任务，但要把connect()里的sync().await()替换为异步回调：

     public void reconnectAttempt() {
         bootstrap.connect(hostname, port).addListener(future -> {
             if (!future.isSuccess()) {
                 // 处理连接失败逻辑，比如再次调度重试
                 ctx.executor().schedule(this::reconnectAttempt, 3, TimeUnit.SECONDS);
             }
         });
     }
     

   • 这种方式的好处是利用Netty自身的线程模型，避免额外线程池的维护，但核心是必须保证重连逻辑非阻塞。

重构后的代码 #

@Override
public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
    super.channelUnregistered(ctx);
    ctx.channel().close();
    client.setConnected(false);
    scheduledExecutorService.schedule(this::reconnectAttempt, 3, TimeUnit.SECONDS);
}

内容的提问来源于stack exchange，提问作者user842588