10线程ScheduledExecutorService执行5次Netty心跳后停止问题排查

问题背景 #

我有一个基于ScheduledExecutorService的Netty客户端应用，负责向服务端发送JSON格式心跳包。核心代码如下：

定时任务初始化 #

private final ScheduledExecutorService ses = Executors.newScheduledThreadPool(10);

log.info("Starting up signaling client.");
this.signalingClient = new SignalingClient(host, port, this);

signalingClient.getChannelFuture().addListener(future -> {
    if (future.isSuccess()) {
        log.info("Connection to signaling server completed. Scheduling heartbeat task.");
        ses.scheduleAtFixedRate(() -> signalingClient.sendMessage(testHeartBeat()),
                3L, 30L, TimeUnit.SECONDS);
    } else {
        log.info("Connection to signaling server completed with exception.");
    }
});
} else {
    log.warn("Unable to start Signaling Gateway. Args must match --voice.adapter.endpoint=127.0.0.1:[port]");
}

SignalingClient核心方法 #

// 发送消息方法
public void sendMessage(WrappedMessage msg) {
    if (channel != null && channel.isActive()) {
        log.info("Writing msg to the channel.");
        channel.writeAndFlush(msg);
    } else {
        log.info("Unable to write message to channel.");
    }
}

// 客户端初始化
public SignalingClient(String host, int port, MessageHandler handler) {
    serverAddress = new InetSocketAddress(host, port);

    bootstrap.group(bossGroup)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.SO_KEEPALIVE, true)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) {
                    ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG))
                            .addLast(new MessageDecoder(MessageDecoder.DecodingState.VERSION))
                            .addLast(new MessageEncoder())
                            .addLast(new AdapterClientStreamHandler(handler));
                }
            });

    this.f = doConnect();
    if (f != null) {
        f.syncUninterruptibly();
    }
}

private ChannelFuture doConnect() {
    try {
        f = bootstrap.connect(serverAddress);
        f.addListener(future -> channel = f.channel());
        return f;
    } catch (Exception ex) {
        log.warn("Connection to signaling server failed: {}.", ex.getMessage());
        return null;
    }
}

核心疑问：明明连接健康、心跳能正常交互，为什么定时任务执行5次后应用就停止了？

排查过程与现象 #

1. 添加Future监听器后的日志分析 #

给sendMessage方法添加ChannelFuture监听器和异常捕获后：

public void sendMessage(WrappedMessage msg) {
    if (channel != null && channel.isActive()) {
        log.info("Writing msg to the channel.");
        try {
            channel.writeAndFlush(msg).addListener(f -> {
                if (f.isSuccess()) {
                    log.info("Successfully wrote message to the channel.");
                } else {
                    log.info("Failed to write message to the channel.");
                }
            });
        } catch (Exception e) {
            log.info("Caught exception in writing message: ", e);
        }

    } else {
        log.info("Unable to write message to channel.");
    }
}

日志显示前4次心跳都成功发送并收到响应，但第5次仅打印Writing msg to the channel.，无后续成功/失败日志：

2023-01-09 17:05:13.635-0800  [I] [26] Writing msg to the channel.
2023-01-09 17:05:13.661-0800  [I] [25] Successfully wrote message to the channel.
2023-01-09 17:05:13.698-0800  [I] [25] Beginning message decode.
2023-01-09 17:05:13.699-0800  [I] [25] Finished decoding message.
2023-01-09 17:05:13.700-0800  [I] [25] Received message from v'1' signaling server: Type: 'HEARTBEAT', Data: '{"type":"heartbeat","data":{"version":"","connections":0,"capabilities":{"udp":[{"address":"10.128.32.86","port":44678}],"tcp":[{"address":"10.128.32.86","port":36197}],"dtls":[{"address":"10.128.32.86","port":44215}]},"expire":"1671582427","instanceId":"b7ea0ad2-363c-40e2-ab15-10eeda74350a"}}'.
2023-01-09 17:05:23.625-0800  [I] [26] Writing msg to the channel.
2023-01-09 17:05:23.626-0800  [I] [25] Successfully wrote message to the channel.
2023-01-09 17:05:23.627-0800  [I] [25] Beginning message decode.
2023-01-09 17:05:23.627-0800  [I] [25] Finished decoding message.
2023-01-09 17:05:23.627-0800  [I] [25] Received message from v'1' signaling server: Type: 'HEARTBEAT', Data: '{"type":"heartbeat","data":{"version":"","connections":0,"capabilities":{"udp":[{"address":"10.128.32.86","port":44678}],"tcp":[{"address":"10.128.32.86","port":36197}],"dtls":[{"address":"10.128.32.86","port":44215}]},"expire":"1671582427","instanceId":"b7ea0ad2-363c-40e2-ab15-10eeda74350a"}}'.
2023-01-09 17:05:33.624-0800  [I] [28] Writing msg to the channel.
2023-01-09 17:05:33.626-0800  [I] [25] Successfully wrote message to the channel.
2023-01-09 17:05:33.627-0800  [I] [25] Beginning message decode.
2023-01-09 17:05:33.627-0800  [I] [25] Finished decoding message.
2023-01-09 17:05:33.627-0800  [I] [25] Received message from v'1' signaling server: Type: 'HEARTBEAT', Data: '{"type":"heartbeat","data":{"version":"","connections":0,"capabilities":{"udp":[{"address":"10.128.32.86","port":44678}],"tcp":[{"address":"10.128.32.86","port":36197}],"dtls":[{"address":"10.128.32.86","port":44215}]},"expire":"1671582427","instanceId":"b7ea0ad2-363c-40e2-ab15-10eeda74350a"}}'.
2023-01-09 17:05:43.626-0800  [I] [26] Writing msg to the channel.
2023-01-09 17:05:43.627-0800  [I] [25] Successfully wrote message to the channel.
2023-01-09 17:05:43.628-0800  [I] [25] Beginning message decode.
2023-01-09 17:05:43.628-0800  [I] [25] Finished decoding message.
2023-01-09 17:05:43.628-0800  [I] [25] Received message from v'1' signaling server: Type: 'HEARTBEAT', Data: '{"type":"heartbeat","data":{"version":"","connections":0,"capabilities":{"udp":[{"address":"10.128.32.86","port":44678}],"tcp":[{"address":"10.128.32.86","port":36197}],"dtls":[{"address":"10.128.32.86","port":44215}]},"expire":"1671582427","instanceId":"b7ea0ad2-363c-40e2-ab15-10eeda74350a"}}'.
2023-01-09 17:05:53.625-0800  [I] [29] Writing msg to the channel.

2. 改用Netty通道Executor的尝试 #

在客户端Handler的channelActive方法中改用Netty自带的Executor执行定时任务，结果依然只执行5次就停止：

@Override
public void channelActive(ChannelHandlerContext ctx) {
    log.info("Connection to the signaling adapter successful. Scheduling heartbeat task.");
    ctx.executor().scheduleAtFixedRate(() -> ctx.writeAndFlush(testHeartBeat()), 0L, 60L, TimeUnit.SECONDS);
}

3. 后续发现的关键现象 #

• 提高定时任务执行频率后，心跳能持续发送；
• 直接通过命令行运行应用一切正常，但用ProcessBuilder启动子进程时，应用会在几分钟后停止。

问题原因与解决方案 #

1. 通道空闲超时 #

虽然配置了SO_KEEPALIVE，但TCP默认的Keep-Alive间隔长达2小时，远不足以覆盖30秒心跳间隔后的空闲窗口。Netty服务端或客户端如果未配置IdleStateHandler，通道长时间无读写操作会被自动关闭，导致后续心跳任务无法完成。

解决方案：在客户端Pipeline中添加IdleStateHandler，主动触发心跳：

// 在初始化Channel时添加IdleStateHandler
ch.pipeline().addLast(new IdleStateHandler(0, 25, 0, TimeUnit.SECONDS)) // 25秒未写触发写空闲
.addLast(new LoggingHandler(LogLevel.DEBUG))
.addLast(new MessageDecoder(MessageDecoder.DecodingState.VERSION))
.addLast(new MessageEncoder())
.addLast(new AdapterClientStreamHandler(handler));

然后在自定义Handler中处理空闲事件：

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
    if (evt instanceof IdleStateEvent) {
        IdleStateEvent event = (IdleStateEvent) evt;
        if (event.state() == IdleState.WRITER_IDLE) {
            ctx.writeAndFlush(testHeartBeat());
        }
    }
    super.userEventTriggered(ctx, evt);
}

这种方式和通道生命周期绑定，比手动定时任务更可靠。

2. ProcessBuilder子进程阻塞问题 #

直接命令行运行正常，但ProcessBuilder启动后停止，核心原因是子进程的标准输入输出缓冲区未被读取，当缓冲区满时，子进程会被阻塞直至停止。

解决方案：必须读取子进程的输出和错误流，或重定向到文件：

// 方式1：启动线程读取流
Process process = new ProcessBuilder("java", "-jar", "your-app.jar").start();

// 读取标准输出
new Thread(() -> {
    try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream()))) {
        String line;
        while ((line = reader.readLine()) != null) {
            // 处理或打印输出
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
}).start();

// 读取错误输出
new Thread(() -> {
    try (BufferedReader reader = new BufferedReader(new InputStreamReader(process.getErrorStream()))) {
        String line;
        while ((line = reader.readLine()) != null) {
            // 处理或打印错误输出
        }
    } catch (IOException e) {
        e.printStackTrace();
    }
}).start();

// 方式2：重定向到文件
ProcessBuilder pb = new ProcessBuilder("java", "-jar", "your-app.jar");
pb.redirectOutput(new File("app.log"));
pb.redirectError(new File("app-error.log"));
Process process = pb.start();

3. 定时任务异常导致终止 #

ScheduledExecutorService的scheduleAtFixedRate方法，如果任务抛出未捕获的异常，会自动取消后续任务执行。检查testHeartBeat()或sendMessage是否存在未处理的异常。

解决方案：给定时任务添加全局异常捕获：

ses.scheduleAtFixedRate(() -> {
    try {
        signalingClient.sendMessage(testHeartBeat());
    } catch (Exception e) {
        log.error("Heartbeat task failed", e);
    }
}, 3L, 30L, TimeUnit.SECONDS);

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