本节目标：把 6.2 的 Reactor 循环补成可跨线程提交任务、可挂定时器、可观测阻塞的调度器。
适用版本：JDK 25（Temurin 25.0.5+8）；Selector、ConcurrentLinkedQueue、PriorityQueue。
承接的问题
11.3 的心跳广播把两个新需求推到了事件循环面前。第一，心跳每 30 秒要扫一遍会话表，给每个 WebSocketSession 入队一个 ping 帧；第二，业务线程处理完请求后，要把响应交回某条连接的写队列。6.2 的 EventLoop 只有 select()、runTasks() 和 processSelectedKeys()，任务队列还是 ArrayDeque，execute() 默认只有循环线程自己调用；select() 也没有超时。11.3 用独立的 ScheduledExecutorService 触发心跳，定时线程和连接状态分属两条线程，靠「只调用入队方法」这条纪律维持边界。
这三个缺口对应本节的三块内容：任务队列要从单生产者改成多生产者单消费者（MPSC），wakeup 要从「停机时叫一下」变成「每次跨线程提交都叫，但能合并就合并」，定时任务要能在循环线程上按截止时间执行，把 11.3 的定时线程收进来，减少一次跨线程边界。
12 章把 6.2 的 EventLoop 类拆成两部分：EventLoopImpl 只负责调度（selector、任务、定时器），register、listen 和 Connection/Handler 的分发留在连接注册层，通过构造器传入的 Consumer<SelectionKey> 接进来。对外行为不变，循环本身变得可以单独测试。
本机口径：Apple M1 Pro，10 核，32 GB，macOS 26.3 (25D125)；Temurin 25.0.5+8。下面标注「本机输出」的数字都来自这台机器。
任务队列：多生产者、单消费者
事件循环的任务队列有三个生产者：业务线程池（12.2）把响应写回、停机钩子提交收尾任务、定时器回调里再提交后续任务；消费者只有一个，就是循环线程。ArrayDeque 只在单线程下安全，多个业务线程同时 add 会破坏内部数组，轻则丢任务，重则抛 ArrayIndexOutOfBoundsException。这里换成 ConcurrentLinkedQueue，它无锁、弱一致，正好匹配「多线程入队、单线程出队」。
execute 的契约要写死：只入队，不内联执行，即使调用者已经在循环线程上。内联执行会破坏 FIFO：一个在循环线程里递归提交的任务会插到已经排队的任务前面；递归层数深时还会把栈打穿。循环线程需要立即生效的操作只有 interestOps 和注册，它们本来就是同步的循环本地状态，不需要经过队列。
private final Queue<Runnable> tasks = new ConcurrentLinkedQueue<>();
private volatile boolean running = true;
public void execute(Runnable task) {
if (task == null) throw new NullPointerException("task");
if (!running) return;
tasks.add(task);
wakeup();
}
public boolean inLoopThread() {
return Thread.currentThread() == loopThread;
}
running 必须是 volatile：停机钩子从另一个线程写 false，循环线程要立刻看见。execute 在停机后静默丢弃任务是刻意的，调用方要么在提交前检查，要么接受「关闭窗口内的任务可能不执行」；14.3 做优雅关闭时会把「停止接收、排空队列、再退出」拆成三个阶段，不靠一个布尔值兜底。队列本身无界，队列里只该放短任务；可能超过毫秒级的动作应去 12.2 的业务线程池，而不是给队列加容量后阻塞生产者。
wakeup：叫醒 select，但别为每个任务叫一次
selector.select() 在没有任何就绪事件时会一直阻塞。业务线程往队列里放了任务，如果不叫醒 selector，任务要等到下一次网络事件才执行；空闲连接上可能几分钟都没有事件。Selector.wakeup() 的作用是让当前或下一次 select() 立即返回，它带有粘性：如果调用时没有 select() 在阻塞，效果保留到下一次。
最直接的写法是每次 execute 都调用 selector.wakeup()。任务少时没问题，高并发下每个响应一次系统调用，循环会被大量空唤醒打断。用一个 AtomicBoolean 合并：只有从「没有待处理唤醒」变成「有待处理唤醒」的那一次才真正调用 wakeup()。
private final AtomicBoolean wakeupPending = new AtomicBoolean();
private void wakeup() {
if (wakeupPending.compareAndSet(false, true)) {
selector.wakeup();
}
}
// run() 里 select 返回之后：
wakeupPending.set(false);
runTasks();
顺序很关键。select() 返回后先把标志清掉，再跑任务；这样在跑任务期间新提交的任务会看到 false，重新触发一次 wakeup()，下一次 select() 不会被漏掉。反过来，如果先跑任务再清标志，任务执行期间提交的任务可能看到 true 而不唤醒，循环随后清掉标志进入 select()，任务就睡在队列里了。
wakeup() 只让 select() 提前返回，不承诺任务先于就绪事件执行。run() 里固定先 runTasks()、再 runTimers()、最后 dispatchSelected()，所以跨线程提交的「关闭连接」「写响应」会排在本次网络事件之前。这个顺序是有意的：业务线程已经等了一轮，先让它落地，减少一次 select 往返。
定时任务：最小堆加 select 超时
定时器可以放在 ScheduledExecutorService，11.3 就是这么做的；但它的回调在另一个线程执行，只能调用 sendPing() 这类入队方法。一旦回调要读会话表、改连接状态，就回到了 6.2 说的跨线程修改问题；如果每次都 execute 回循环，又多一次队列跳转和延迟。12 章把定时器收进 EventLoopImpl，让「什么时候该跑」和「真正执行」在同一条线程上。
数据结构用 PriorityQueue，按 System.nanoTime() 的绝对截止时间排序。schedule 可能从任意线程调用，堆的读写都放在同一把锁里；循环线程取堆顶时也拿这把锁。nanoTime() 是单调时钟，不受系统时间调整影响；用 currentTimeMillis() 算间隔，遇到 NTP 回拨会把定时器推迟或提前。
private long nextTimeoutMillis() {
synchronized (timerLock) {
TimerTask head = timers.peek();
if (head == null) return -1L;
long delay = head.deadlineNanos - System.nanoTime();
if (delay <= 0L) return 0L;
return Math.max(1L, TimeUnit.NANOSECONDS.toMillis(delay));
}
}
返回值有三种含义：-1 表示没有定时器，调用无参 select() 一直阻塞；0 表示堆顶已到期，用 select(0) 立即轮询；正数表示最多睡这么久。循环每轮都重新计算，因为 runTasks() 可能跑了很久，进 select() 时原来的截止时间已经过了。
固定频率任务在每次执行后把 deadlineNanos 加上周期，而不是用「执行结束时间 + 周期」。前者保持长期频率稳定，后者会随任务耗时漂移。如果一次执行已经落后超过一个周期，直接把下一次截止时间推到「现在 + 周期」，跳过积压的 tick；否则一次长阻塞之后会连续补跑多次心跳，把循环再次压住。
TimerHandle.cancel() 只把标志置 true，不从堆里删除；被取消的任务到期后被跳过。定时器数量在几百以内时这样最简单；如果大量任务频繁取消，堆里会堆积无效节点，需要换成时间轮或带索引的堆。
一次阻塞，所有定时器一起迟到
事件循环的串行性意味着：一个任务阻塞多久，所有连接和所有定时器就等多久。下面这个实验在 200ms 处挂一个心跳定时器，100ms 时提交一个睡眠 500ms 的任务，沿用上面的编译命令运行：
loop.schedule(() -> log("heartbeat due at 200ms"), 200);
Thread.sleep(100);
loop.execute(() -> {
log("blocking task start");
Thread.sleep(500);
log("blocking task end");
});
本机输出（单次运行，未做统计）：
136 ms blocking task start
653 ms blocking task end
654 ms heartbeat due at 200ms
心跳的截止时间是 200ms，实际在 654ms 才执行，正好被 500ms 的阻塞任务挡在身后。真实服务里这段阻塞可能来自 JDBC 查询、文件读写、InetAddress.getByName() 的 DNS 解析、Future.get()、锁竞争，甚至一次很大的 JSON 序列化。它们共同的特征是：在事件循环线程上等待外部资源。
检测办法不是靠代码评审，而是给每个任务计时。超过阈值就记一条 WARN，带上任务类和耗时；不要中断任务，中断只会让状态更难恢复。
private void runOne(Runnable task) {
long start = System.nanoTime();
try { task.run(); }
finally {
long ms = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
if (ms > 50L) System.getLogger("linkd").warning("slow task " + task.getClass() + ": " + ms + " ms");
}
}
在 linkd 里，心跳和广播都应该拆成短任务。心跳扫描只做「遍历会话、给到期连接 enqueue ping 帧」，编码可以提前做好；广播一万条会话时，按 256 条一批用 schedule(..., 0) 分片，每批之间让循环处理一次读写。这样单次占用循环的时间有上限，心跳不会因为一次全量广播而整体推迟。
在 linkd 里落地：EventLoopImpl
下面按「循环与提交」和「任务与定时器执行」两段看同一个类，包名 com.plume.linkd.core。
package com.plume.linkd.core;
import java.io.IOException;
import java.nio.channels.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
public final class EventLoopImpl implements Runnable {
private final Selector selector;
private final Consumer<SelectionKey> keyHandler;
private final Queue<Runnable> tasks = new ConcurrentLinkedQueue<>();
private final PriorityQueue<TimerTask> timers = new PriorityQueue<>();
private final Object timerLock = new Object();
private final AtomicBoolean wakeupPending = new AtomicBoolean();
private volatile Thread loopThread;
private volatile boolean running = true;
public EventLoopImpl(Consumer<SelectionKey> keyHandler) throws IOException {
this.selector = Selector.open();
this.keyHandler = keyHandler;
}
public void execute(Runnable task) {
if (task == null) throw new NullPointerException("task");
if (!running) return;
tasks.add(task);
wakeup();
}
public TimerHandle schedule(Runnable task, long delayMillis) {
return schedule(task, delayMillis, 0L);
}
public TimerHandle scheduleAtFixedRate(Runnable task, long initialDelayMillis, long periodMillis) {
if (periodMillis <= 0L) throw new IllegalArgumentException("period must be positive");
return schedule(task, initialDelayMillis, periodMillis);
}
private TimerHandle schedule(Runnable task, long delayMillis, long periodMillis) {
if (!running) throw new IllegalStateException("event loop is shut down");
TimerHandle handle = new TimerHandle();
long deadline = System.nanoTime()
+ TimeUnit.MILLISECONDS.toNanos(Math.max(0L, delayMillis));
synchronized (timerLock) {
timers.add(new TimerTask(handle, task, deadline,
TimeUnit.MILLISECONDS.toNanos(periodMillis)));
}
wakeup();
return handle;
}
public boolean inLoopThread() {
return Thread.currentThread() == loopThread;
}
@Override
public void run() {
loopThread = Thread.currentThread();
try {
while (running) {
long timeout = nextTimeoutMillis();
int ready = timeout < 0L ? selector.select() : selector.select(timeout);
wakeupPending.set(false);
runTasks();
runTimers();
if (ready > 0) dispatchSelected();
}
} catch (IOException e) {
if (running) System.getLogger("linkd").log(System.Logger.Level.ERROR, "selector failed", e);
} finally {
try { selector.close(); } catch (IOException ignored) { }
}
}
public void shutdown() {
running = false;
wakeup();
}
private void wakeup() {
if (wakeupPending.compareAndSet(false, true)) selector.wakeup();
}
private long nextTimeoutMillis() {
synchronized (timerLock) {
TimerTask head = timers.peek();
if (head == null) return -1L;
long delay = head.deadlineNanos - System.nanoTime();
if (delay <= 0L) return 0L;
return Math.max(1L, TimeUnit.NANOSECONDS.toMillis(delay));
}
}
第二段是执行侧。runTasks 把队列一次排空；runTimers 只执行已到期的堆顶，固定频率任务重新入堆；dispatchSelected 遍历就绪 key，交给构造器传入的 keyHandler。
private void runTasks() {
Runnable task;
while ((task = tasks.poll()) != null) {
try {
task.run();
} catch (RuntimeException e) {
System.getLogger("linkd").log(System.Logger.Level.WARNING, "task failed", e);
}
}
}
private void runTimers() {
while (true) {
TimerTask timer;
synchronized (timerLock) {
timer = timers.peek();
if (timer == null || timer.deadlineNanos > System.nanoTime()) return;
timers.poll();
}
if (timer.handle.cancelled) continue;
try {
timer.task.run();
} catch (RuntimeException e) {
System.getLogger("linkd").log(System.Logger.Level.WARNING, "timer failed", e);
}
if (timer.periodNanos > 0L && !timer.handle.cancelled) {
synchronized (timerLock) {
timer.deadlineNanos += timer.periodNanos;
long now = System.nanoTime();
if (timer.deadlineNanos <= now) timer.deadlineNanos = now + timer.periodNanos;
timers.add(timer);
}
}
}
}
private void dispatchSelected() {
var iterator = selector.selectedKeys().iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
if (!key.isValid()) continue;
try { keyHandler.accept(key); }
catch (RuntimeException e) {
key.cancel();
try { key.channel().close(); } catch (IOException ignored) { }
}
}
}
public static final class TimerHandle {
private volatile boolean cancelled;
public void cancel() { cancelled = true; }
}
private static final class TimerTask implements Comparable<TimerTask> {
private final TimerHandle handle;
private final Runnable task;
private final long periodNanos;
private long deadlineNanos;
private TimerTask(TimerHandle h, Runnable t, long deadline, long period) {
handle = h; task = t; deadlineNanos = deadline; periodNanos = period;
}
@Override
public int compareTo(TimerTask other) {
return Long.compare(deadlineNanos, other.deadlineNanos);
}
}
}
装配时，LinkdMain 创建 EventLoopImpl，把连接层的分发 lambda 传进去；listen、register、Connection 的构造仍由连接注册层完成。停机钩子调用 shutdown()，它只置标志并 wakeup()，selector.close() 留给 run() 的 finally，避免在另一个线程关闭正在 select() 的 selector。
任务和定时器都在循环线程上执行，所以 Connection.outbound、SelectionKey 的 interestOps 仍然不需要锁。跨线程来的只有「提交」这个动作，队列和 AtomicBoolean 负责把动作安全地交给循环线程。
运行与行为验证
把类放进 linkd-core/src/main/java/com/plume/linkd/core/。验证程序在 100ms 时提交一个普通任务，150ms 时挂一次性定时器，200ms 时启动每 200ms 一次的心跳并在第三次取消，900ms 后关停：
export JAVA_HOME=~/.jdks/jdk-25.0.5+8/Contents/Home
javac --release 25 -d out $(find src -name '*.java')
java -cp out com.plume.linkd.core.EventLoopDemo
本机输出（单次运行，未做统计；毫秒数含 JVM 启动和调度噪声）。验证程序在类上另加了三个 LongAdder 计数（完成的任务、完成的定时器、真实 wakeup 次数），只用于这组输出：
129 ms execute task, inLoop=true
282 ms one-shot timer, inLoop=true
340 ms heartbeat tick 1, inLoop=true
538 ms heartbeat tick 2, inLoop=true
737 ms heartbeat tick 3, inLoop=true
1060 ms stats tasks=1 timers=4 pendingTasks=0 pendingTimers=0 wakeups=2
1061 ms loopAlive=false
四个回调全部打印 inLoop=true，说明任务、一次性定时器和固定频率心跳都落在 linkd-io-0 上。timers=4 是一次性定时器加三次心跳；wakeups=2 不是提交次数（三次），而是合并后的真实唤醒次数：第一次提交与紧随其后的定时器注册落在同一个唤醒窗口里，只叫醒了一次 select。毫秒数和期望值有几毫秒到几十毫秒的偏差，来自线程启动、类加载和 select 的超时粒度，不是定时器精度承诺。
常见错误与边界
错误一：用 ArrayDeque 当跨线程任务队列。 业务线程和循环线程同时 add/poll，内部数组会损坏。本机曾出现任务莫名丢失和 ArrayIndexOutOfBoundsException；换成 ConcurrentLinkedQueue 后消失。
错误二：execute 在循环线程上直接 task.run()。 看起来省了一次入队，实际破坏了 FIFO：递归提交的任务会插队；连续递归几十万层会 StackOverflowError。统一入队，循环下一轮再跑。
错误三：提交任务后忘记 wakeup()。 空闲连接上没有网络事件，select() 会一直睡；任务要等到下一个连接发数据才执行。单线程版 execute 只被循环线程调用时不容易暴露，12.2 接入业务线程后立刻变成「响应偶尔慢几秒」。
错误四：每次 execute 都无条件 wakeup()。 功能正确，但高并发下每个响应一次系统调用，循环被空唤醒打满。用 AtomicBoolean 合并；注意必须在 select() 返回后、跑任务前清标志，顺序反了会漏任务。
错误五：定时器回调里抛异常不处理。 这里的 runTimers 捕获 RuntimeException 并继续；如果换成 ScheduledExecutorService，一次异常会让后续执行被取消，心跳静默停止。定时任务要么自己捕获，要么由调度器兜底。
边界：定时器不承诺准时。 它只做到「不早于截止时间执行」，前提是循环没有被长任务占住。要更准的调度，需要把长任务移出循环，或者用独立的调度线程只做触发、把执行 execute 回循环。
边界：shutdown() 不排空任务队列。 它只置 running=false 并唤醒；队列里未执行的任务被丢弃。需要「排空再关」的场景由 14.3 的三阶段关闭处理，不在调度器里硬编码。
小结
- 任务队列是多生产者单消费者，用
ConcurrentLinkedQueue；execute只入队，不在调用线程内联执行。 wakeup()用AtomicBoolean合并，select()返回后先清标志再跑任务，避免漏掉执行期间提交的任务。- 定时任务用
PriorityQueue按nanoTime()截止时间排序，select的超时取堆顶剩余时间；固定频率任务重排截止时间，落后太多就跳过积压 tick。 - 循环顺序是任务、定时器、就绪 key；跨线程提交的写回和关闭排在本次网络事件之前。
- 事件循环上任何阻塞都会让所有连接和定时器一起等；给任务计时，超过阈值记 WARN，把长任务移到业务线程池。
EventLoopImpl只负责调度，连接注册和分发通过Consumer<SelectionKey>注入，循环因此可以单独测试。
任务队列解决了「谁来跑」，但没有解决「跑多久」。下一节把阻塞业务移出循环：业务线程池开多大、队列多长、拒绝后回什么，以及虚拟线程在这个位置上能替代什么、不能替代什么。
阅读导航：上一节：11.3 心跳、关闭与广播 · 下一节：12.2 业务线程池与阻塞隔离 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。