《Netty 高性能网络编程》3.2 任务队列与定时任务

本节目标：把任务提交、定时、重连退避和心跳调度放到正确的线程上，并知道它们什么时候会迟到或停止。
适用版本：JDK 25 + Netty 4.2.19.Final（netty-common、netty-transport 模块）。

承接的问题

3.1 确认了一条连接的所有事件都在同一个 EventLoop 上串行执行。接下来 linkd-v2 有三个跨线程需求：

• 客户端连接失败后要按退避重试，由 ReconnectHandler 负责；
• WebSocket 会话每 30 秒要发一次 ping，由 WebSocketHub 负责；
• 网关转发 libraryd 时要设置超时，不能在等待上无限挂住。

最初的做法是每个连接 new 一个 ScheduledExecutorService，回调里再调用 Channel 的方法。连接数上千后线程数跟着上千，而且回调线程和 IO 线程不是同一条，连接状态要靠锁或队列传递。实际上 EventLoop 本身就是一个单线程的 ScheduledExecutorService，每一条连接已经有一条专属线程，不需要再为它创建调度器。

这一节要回答的是：任务提交到 EventLoop 后按什么顺序执行、定时任务为什么会被普通任务推迟、ChannelFuture 的监听器在哪个线程跑、GlobalEventExecutor 和 DefaultEventExecutorGroup 分别该用在哪里，以及心跳任务写成定时器后为什么会悄悄停掉。

本机口径：Apple M1 Pro，10 核，32 GB，macOS 26.3 (25D125)；Temurin 25.0.5+8，Netty 4.2.19.Final。下面标注「本机输出」的内容都来自这台机器，单次运行，未做统计。

EventLoop 就是一个单线程 ScheduledExecutorService

EventLoop 是一条单线程的 EventExecutor；EventExecutor 所属的 EventExecutorGroup 实现了 ScheduledExecutorService，单个 EventLoop 也对外提供同样的调度方法。常用的入口有四个：

var loop = channel.eventLoop();
loop.execute(() -> System.out.println("task"));
Future<?> oneShot = loop.schedule(
        () -> System.out.println("one shot"), 100, TimeUnit.MILLISECONDS);
Future<?> fixedRate = loop.scheduleAtFixedRate(
        () -> System.out.println("tick"), 0, 50, TimeUnit.MILLISECONDS);
Future<?> fixedDelay = loop.scheduleWithFixedDelay(
        () -> System.out.println("tick"), 0, 50, TimeUnit.MILLISECONDS);

execute 和 submit 是普通任务，schedule 是一次性定时任务，scheduleAtFixedRate 和 scheduleWithFixedDelay 是周期任务。四个方法都可以从任意线程调用，任务最终都在这个 EventLoop 的线程上执行。

提交规则有两条。第一，从其他线程调用 execute，任务进入队列并唤醒 selector，不会在调用线程上内联执行；即使调用线程碰巧就是 EventLoop 线程，任务也会排队，等当前回调返回后才执行。第二，同一个线程连续提交的任务保持先进先出；不同线程并发提交时没有全局顺序，只能做到每个任务最终被执行。

boolean inLoop = loop.inEventLoop();

inEventLoop() 用来判断当前线程是否属于这个 EventLoop。它适合做断言和分支，不适合作为「可以安全访问状态」的唯一依据：handler 被添加到 DefaultEventExecutorGroup 后，回调线程不再等于 channel.eventLoop()，但连接状态依然只被那一条业务线程访问，判断条件要跟着改。

定时任务的返回值是 Netty 的 Future，支持 cancel、addListener 和 isDone。在 EventLoop 线程里不要调用 get() 或 sync() 等待它完成，原因在常见错误里展开。

任务与定时器在同一条线程上排队

EventLoop 的线程既处理 IO 事件，也处理任务队列和定时器。三者没有独立的线程，所以一个长任务会同时推迟后两者。下面这段探针在一条已连接的 Channel 上提交四个动作：

long t0 = System.nanoTime();
var loop = channel.eventLoop();
loop.execute(() -> log("execute", t0));
loop.schedule(() -> log("timer B due at 100ms", t0), 100, TimeUnit.MILLISECONDS);
loop.schedule(() -> {
    log("timer C due at 150ms, sleep 300ms", t0);
    Thread.sleep(300);
    log("timer C sleep done", t0);
}, 150, TimeUnit.MILLISECONDS);
loop.schedule(() -> log("timer D due at 200ms", t0), 200, TimeUnit.MILLISECONDS);
channel.closeFuture().addListener(f -> log("close listener", t0));

本机输出（单次运行，未做统计；毫秒数含线程启动和调度噪声）：

 226 ms  main thread name=main
 238 ms  execute from main, inEventLoop=true
 338 ms  timer B due at 100ms
 388 ms  timer C due at 150ms, sleep 300ms
 689 ms  timer C sleep done
 689 ms  timer D due at 200ms
 945 ms  close listener, inEventLoop=true

四个回调全部打印 inEventLoop=true，说明它们都在同一条 nioEventLoopGroup 线程上执行。B 在提交后约 100 毫秒执行，C 在约 150 毫秒执行，符合预期。D 的截止时间是提交后 200 毫秒，也就是大约 438 毫秒，实际到 689 毫秒才执行，因为 C 在 388 毫秒开始睡了 300 毫秒，D 只能排在它后面。

这就是「任务在 IO 线程执行」的核心风险：Thread.sleep(300)、JDBC 查询、同步 HTTP 调用、文件读写、Future.get()，任何一种阻塞都会把同一 EventLoop 上的读写和定时任务一起推迟。本机这次 300 毫秒的睡眠让 D 迟到了约 251 毫秒；如果在生产环境里每个请求都做一次 50 毫秒的数据库查询，同一 EventLoop 上其他连接的心跳和超时都会跟着漂移。

ChannelFuture 的监听器也遵守同样的规则。addListener 注册的回调在 Channel 所属的 EventLoop 上执行，不在注册监听器的线程上执行。上面输出里的 close listener 在 945 毫秒打印，线程是 EventLoop，不是 main。

固定频率与固定延迟

scheduleAtFixedRate 按「上一次的截止时间 + 周期」计算下一次，任务执行时间超过周期时会尽快补跑；scheduleWithFixedDelay 按「上一次执行结束 + 延迟」计算，不会补跑。本机在 3.3 的阻塞探针里看到过固定频率的补跑现象：50 毫秒一次的心跳在阻塞期间只出现了一次，阻塞结束后同一毫秒连续打印了多次 tick。这个行为可以用来保持长期频率，但不适合用来统计「心跳次数」或驱动有副作用的业务，因为一次迟到会带来一串补偿执行。

心跳任务通常很短，用固定频率没有问题；如果任务本身可能超过周期，用固定延迟更稳，至少不会在恢复后瞬间堆出一批执行。

任务抛异常之后

普通任务抛异常不会杀死 EventLoop。本机输出（节选）：

警告: A task raised an exception. Task: demo.TaskExceptionProbe$$Lambda/0x0000180001065c08@719eee27
java.lang.IllegalStateException: boom-execute
second task ran on nioEventLoopGroup-2-1

Netty 捕获异常并记录一条 WARN，后续任务继续执行。周期任务不一样：本机让第一次 tick 抛异常，300 毫秒后计数仍然是 1：

tick 1 on nioEventLoopGroup-2-1
runs after 300ms = 1

后续 tick 全部停止，而且没有明显的业务日志。如果心跳任务体里有一次空指针，心跳就会静默消失，连接看起来还活着，直到空闲超时被对端关闭。周期任务的函数体必须自己 try/catch，这是硬要求，不是风格建议。

用 ChannelFuture 回调调度重连

ReconnectHandler 的重连逻辑适合用 ChannelFuture.addListener 加 EventLoop.schedule 组合：连接失败的回调本来就在 EventLoop 上，安排下一次重试不需要跨线程。

package com.plume.linkd.v2.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.util.concurrent.Future;

import java.util.concurrent.TimeUnit;

public final class ReconnectHandler {
    private final Bootstrap bootstrap;
    private final String host;
    private final int port;
    private volatile boolean shutdown;
    private volatile Future<?> pending;
    private int attempts;

    public ReconnectHandler(Bootstrap bootstrap, String host, int port) {
        this.bootstrap = bootstrap;
        this.host = host;
        this.port = port;
    }

    public void reconnect() {
        if (shutdown) return;
        ChannelFuture connect = bootstrap.connect(host, port);
        connect.addListener(f -> {
            if (f.isSuccess()) {
                attempts = 0;
                pending = null;
                return;
            }
            if (shutdown) return;
            long delay = 100L << Math.min(attempts, 5);
            attempts++;
            pending = connect.channel().eventLoop()
                    .schedule(this::reconnect, delay, TimeUnit.MILLISECONDS);
        });
    }

    public void shutdown() {
        shutdown = true;
        Future<?> task = pending;
        if (task != null) {
            task.cancel(false);
        }
    }
}

attempts 只在 EventLoop 上读写，不需要 volatile；shutdown 和 pending 会被外部线程访问。每次失败只保留一个 pending 任务，成功后清空，避免多次失败叠加出多条重试链。

本机探针连一个没有监听的端口，退避序列是 100、200、400 毫秒：

  94 ms  attempt 1 on main
 203 ms  failed, retry in 100 ms
 305 ms  attempt 2 on nioEventLoopGroup-2-1
 306 ms  failed, retry in 200 ms
 507 ms  attempt 3 on nioEventLoopGroup-2-1
 508 ms  failed, retry in 400 ms
 909 ms  attempt 4 on nioEventLoopGroup-2-1
 913 ms  give up after 4 attempts

第一次调用来自 main，失败回调在 EventLoop 上，之后的重试也都在 nioEventLoopGroup-2-1 上。时间点 305、507、909 与 100、200、400 的间隔吻合，误差来自连接失败检测和调度粒度。

真实客户端还要加两个东西。第一是抖动：所有实例在同一时刻断线，固定退避会让它们在同一毫秒重连，给服务端制造尖峰；在 delay 上乘一个 0.8 到 1.2 的随机系数。第二是上限：指数退避要封顶，否则进程运行一天后延迟会大到没有意义。ReconnectHandler 里用 Math.min(attempts, 5) 把上限压在 3.2 秒，实际项目按下游恢复时间调整。

不要在失败回调里写 Thread.sleep(delay) 然后递归调用。回调在 EventLoop 上，睡眠会阻塞整个线程，而且递归会让调用栈越来越深。用 schedule 把下一次尝试变成一个新的任务。

GlobalEventExecutor 与 DefaultEventExecutorGroup

GlobalEventExecutor 是 Netty 内部的单线程兜底执行器，DefaultEventExecutorGroup 是给 handler 用的业务线程组。两者名字都带 executor，定位完全不同。

GlobalEventExecutor 的线程名是 globalEventExecutor-1-1，本机输出：

globalEventExecutor-1-1 virtual=false
defaultEventExecutorGroup-2-1 virtual=false
defaultEventExecutorGroup-2-2 virtual=false

它只有一个线程，启动后跟随 JVM。实际包名是 io.netty.util.concurrent.GlobalEventExecutor，在 netty-common 模块里。调用 shutdownGracefully() 返回的是一个失败的 Future，.sync() 会抛出：

io.netty.util.concurrent.GlobalEventExecutor$StacklessUnsupportedOperationException: null

所以不要把它当成自己拥有的线程池去管理，也不要把业务任务放进去。它适合极短的通知类动作，例如 DefaultChannelGroup 需要一个执行器来串行化关闭事件。

DefaultEventExecutorGroup 的用法是在 pipeline 里给 handler 指定执行器：

DefaultEventExecutorGroup business = new DefaultEventExecutorGroup(2);
channel.pipeline().addLast(business, new SlowHandler());

此后 SlowHandler 的 channelRead、channelActive 等回调都在 defaultEventExecutorGroup 的线程上执行，而不是 nioEventLoopGroup 线程。本机输出里 handler 打印的线程是 defaultEventExecutorGroup-2-1 和 -2-2。它解决了「阻塞 handler 拖住 IO 线程」的问题，但没有解决容量问题：两条业务线程同一时刻只能跑两个阻塞任务，其余任务在队列里等。第 3.3 节会用四条连接压两条业务线程，第二波请求晚了约 205 毫秒才被处理。

心跳定时任务的正确与错误写法

先看四种错误写法，它们的共同点是把定时器当成了没有代价的闹钟。

反例一：每条连接在 channelActive 里挂一个周期任务，关闭时忘记取消。 任务持有 Channel 引用，连接关闭后仍每 30 秒执行一次，越积越多。周期任务必须在 channelInactive 或 handlerRemoved 里 cancel。

反例二：在任务里直接查数据库或调用下游 HTTP。 这会阻塞定时器所在的 EventLoop；如果任务挂在某条连接的 EventLoop 上，同一线程上的所有连接一起卡住。任务只应该做「遍历会话、入队 ping 帧」这种内存操作。

反例三：任务体不捕获异常。 如上一节实测，周期任务第一次抛异常后就不再执行，心跳静默停止。任务体必须自己 try/catch，记录日志后继续下一轮。

反例四：用固定频率跑一个可能超过周期的任务。 阻塞结束后会连续补跑，瞬间发出一批 ping，反而制造流量尖峰。心跳发送要短，或者在任务里检查「上一轮是否还在执行」。

正确写法是全局一个扫描任务，只做入队，异常自己兜住，关闭时取消：

package com.plume.linkd.v2.server;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.concurrent.Future;

import java.util.concurrent.TimeUnit;

public final class WebSocketHub extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    private final ChannelGroup sessions;
    private volatile Future<?> heartbeat;

    public WebSocketHub(ChannelGroup sessions) {
        this.sessions = sessions;
    }

    public void startHeartbeat(EventExecutorGroup group) {
        heartbeat = group.next().scheduleAtFixedRate(() -> {
            try {
                sessions.writeAndFlush(new PingWebSocketFrame());
            } catch (Throwable t) {
                System.getLogger("linkd").log(System.Logger.Level.WARNING,
                        "heartbeat failed", t);
            }
        }, 30, 30, TimeUnit.SECONDS);
    }

    public void stopHeartbeat() {
        Future<?> task = heartbeat;
        if (task != null) {
            task.cancel(false);
            heartbeat = null;
        }
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) {
        sessions.writeAndFlush(new TextWebSocketFrame(frame.text()));
    }
}

sessions.writeAndFlush 是线程安全的，内部会把写操作排到每个 Channel 自己的 EventLoop，所以心跳任务本身只做遍历和入队。会话数量很大时，一次遍历仍然可能占用定时器线程几十毫秒；10.3 会把广播按批分片，这里先做到任务里没有阻塞调用。

startHeartbeat 接收 EventExecutorGroup 而不是具体的 EventLoop，调用方传入 worker 组即可，由 next() 选一条线程承载定时器。整个进程只跑一个心跳任务，比每条连接一个任务少了 N 倍的调度开销和取消逻辑。

在 linkd-v2 里落地

LinkdV2Main 启动时创建 WebSocketHub，把心跳挂在 worker 组上；关闭时先停心跳，再关会话和监听 Channel：

public void start() throws InterruptedException {
    servers.add(bind(9090));
    servers.add(bind(9093));
    hub.startHeartbeat(worker);
}

public void shutdown() {
    hub.stopHeartbeat();
    for (Channel server : servers) {
        server.close();
    }
    boss.shutdownGracefully();
    worker.shutdownGracefully();
}

会话组的关闭放在 10.3 的 WebSocket 关闭流程里，这里只要求心跳任务先停，避免它在会话组关闭后继续遍历。ChannelGroup 位于 io.netty.channel.group 包，属于 netty-transport 模块。

ReconnectHandler 放在客户端侧，RpcClient 持有一个实例。连接成功时 attempts 归零，连接断开时由 channelInactive 触发 reconnect()。重连任务和心跳任务都不提交到 GlobalEventExecutor，因为它们是业务任务，需要可控的生命周期和独立的线程归属。

如果会话数量大到单个心跳任务无法在一毫秒内完成，可以按 EventLoop 分片：为每个 worker 线程各挂一个任务，每个任务只遍历注册在自己 EventLoop 上的 Channel。分片后单次任务的工作量与连接总数解耦，代价是需要维护「Channel 属于哪个 EventLoop」的索引。ConnectionRegistry 可以在 add 时按 channel.eventLoop() 分组，这个索引在第 9 章的写缓冲统计里还会用到。

常见错误与边界

错误一：以为 execute 会内联执行。 在 EventLoop 线程里递归调用 execute 不会立刻执行，而是排队到当前回调之后；如果代码依赖「提交后状态立刻改变」，会读到旧值。需要同步生效的状态修改应该直接改，不要绕任务队列。

错误二：在任务里调用 sync() 或 get()。 本机在 EventLoop 线程里发起连接并 sync()，得到：

io.netty.util.concurrent.BlockingOperationException: DefaultChannelPromise@74640e71(incomplete)

Netty 只能识别「等待的 Future 属于当前 EventLoop」这一种死锁；等待 EventLoopGroup 的终止 Future 不在此列，会直接挂住线程（3.1 已实测）。在 EventLoop 上只用 addListener，不要阻塞等待。

错误三：cancel(false) 能停止正在执行的任务。 不能。cancel(false) 只阻止尚未开始的任务；已经在跑的任务会继续到结束。需要中断语义时用 cancel(true)，但任务体必须正确处理中断，否则只是设置了一个标志。

错误四：把 GlobalEventExecutor 当业务线程池。 它只有一条线程，所有使用者共享；一次阻塞会拖住 Netty 内部依赖它的通知。它的关闭 Future 本身是失败的，调用 .sync() 会抛 StacklessUnsupportedOperationException。

错误五：任务里捕获了 Throwable 却继续抛。 周期任务的异常必须被吞掉并记录；重新抛出等于让后续 tick 全部取消。捕获后至少记一条 WARN，带上任务名和轮次，否则心跳停了没有任何线索。

边界：定时任务不承诺准时。 schedule 的语义是「不早于截止时间」，不是「恰好在截止时间」。EventLoop 被 IO 事件或长任务占用时，定时器只能等。对时间敏感的超时控制，要么把长任务移出 EventLoop，要么接受毫秒级到百毫秒级的偏差。

边界：任务队列没有容量上限。 从外部线程无限制地 execute，队列会一直增长，直到内存耗尽。需要背压时，在提交侧做限流或使用有界队列，不要指望 EventLoop 自己拒绝。

小结

• EventLoop 是单线程 ScheduledExecutorService，execute、schedule、scheduleAtFixedRate 的任务都在同一条线程上执行，与 IO 事件共享时间片。
• 同线程提交保持 FIFO，跨线程提交没有全局顺序；任务永远排队，不会在调用线程内联执行。
• 定时任务只承诺「不早于截止时间」，一个 300 毫秒的阻塞任务能让 200 毫秒到期的任务迟到约 251 毫秒。
• ChannelFuture 的监听器在 Channel 的 EventLoop 上执行，适合直接安排重连退避；本机退避序列 100、200、400 毫秒，重试线程始终是 nioEventLoopGroup-2-1。
• 普通任务抛异常只记 WARN，周期任务抛异常会停止后续执行且没有明显业务日志；周期任务体必须自己 try/catch。
• GlobalEventExecutor 是单线程兜底，不归使用者关闭；DefaultEventExecutorGroup 把 handler 移出 IO 线程，但线程数就是阻塞并发上限。

定时任务解决了「什么时候做」，但没有解决「做多久」。下一节把 JDBC、文件 IO 和同步 HTTP 这类真正的阻塞调用移出 EventLoop，比较业务线程组、独立线程池和虚拟线程三种隔离方式，并说明虚拟线程不能替代容量规划。

阅读导航：上一节：3.1 EventLoopGroup 与线程归属 · 下一节：3.3 阻塞隔离与虚拟线程 。