《Java 网络编程基础》6.3 多线程 Reactor 与主从模型

本节目标：把 accept 与读写拆到不同线程，用 round-robin 把连接固定到 I/O 线程，并说清跨线程提交与可见性边界。
适用版本：JDK 25（Temurin 25.0.5+8）；Selector、ConcurrentLinkedQueue、AtomicInteger。

承接的问题

6.2 的单线程 Reactor 能接住大量空闲连接，但它只用一个线程做 accept、read、write。任何一次慢 handler 都会把后面所有连接排住；机器有 10 个核，事件循环只用了 1 个。

多线程 Reactor 的目标不是「把 selector 放进线程池」，而是把职责拆开：一个 accept 线程只负责收连接，多个 I/O 线程各自持有 selector，负责已建立连接的读写。 连接一旦分配给某个 I/O 线程，就固定在那里，不跨线程迁移。这样每条连接的状态仍然只属于一个线程，6.2 的线程归属规则不用推翻。

本机口径：Apple M1 Pro，10 核，32 GB，macOS 26.3 (25D125)；Temurin 25.0.5+8。

主从模型：boss 只 accept，worker 只读写

linkd-accept (boss)
  Selector: OP_ACCEPT
    dispatch round-robin
      ├─ linkd-io-0  Selector -> Connection A, E
      ├─ linkd-io-1  Selector -> Connection B, F
      ├─ linkd-io-2  Selector -> Connection C, G
      └─ linkd-io-3  Selector -> Connection D, H

boss 线程的 selector 上只注册监听通道，它不做 read/write，也不执行业务逻辑。worker 线程各自 Selector.open()，只处理已建立连接；它们不调用 accept()。这种结构通常叫主从 Reactor：主 reactor 负责 accept，从 reactor 负责 I/O。

不能让多个线程共享一个 selector。Selector 的注册集合和就绪集合不是线程安全的，多个线程同时 select、register、改 interestOps 会把状态搅乱。正确做法是每个 I/O 线程一个 selector，连接按某种规则分配过去。

macOS 上每个 Selector 对应一个 kqueue 实例。本机起 4 个 worker 加 1 个 boss，lsof 能看到 5 个 kqueue 条目：

kqueue entries: 5

这不是「线程越多越好」的证据，只是说明 selector 与内核对象的对应关系。worker 数量应该从 CPU 核数起步，再用压测调整；超过核数后，上下文切换和缓存失效会吃掉收益。

另一种做法是多个线程各自 bind 同一端口并开 SO_REUSEPORT，让内核分流连接。它省掉 accept 线程，但连接分配由内核决定，调试和优雅关闭更复杂；本卷不展开。

跨线程注册：把 register 投递给目标循环

boss 接受连接后，不能直接调用 worker.register(channel, handler)。EventLoop.register 在 6.2 里已经加了线程检查，本机直接调用会得到：

java.lang.IllegalStateException: register must run on the event loop thread

正确做法是把「注册」包装成任务，投递给目标 worker：

private void dispatch(SocketChannel channel, HandlerFactory factory) {
    EventLoop worker = workers[Math.floorMod(next.getAndIncrement(), workers.length)];
    worker.execute(() -> {
        try { worker.register(channel, factory.create(worker)); }
        catch (IOException e) {
            try { channel.close(); } catch (IOException ignored) { }
        }
    });
}

HandlerFactory 是一个很小的函数式接口，让每个连接拿到自己的 handler：

package com.plume.linkd.core;

@FunctionalInterface
public interface HandlerFactory {
    Handler create(EventLoop loop);
}

factory.create(worker) 在 worker 线程上执行，不是在 boss 线程。这样 handler 的构造、字段初始化、对 BufferPool 的取用都发生在目标循环里，不需要额外同步。如果 factory 里访问了跨线程共享的可变对象，那个对象仍然要自己承担线程安全。

execute 必须使用并发队列。6.2 的 ArrayDeque 只允许循环线程自己读写；多线程提交要换成 ConcurrentLinkedQueue：

private final Queue<Runnable> tasks = new ConcurrentLinkedQueue<>();

public void execute(Runnable task) {
    tasks.add(task);
    selector.wakeup();
}

tasks.add 与 worker 线程的 tasks.poll 之间建立 happens-before：任务里对 channel、handler 的初始化，在 poll 之后对 worker 可见。wakeup() 让阻塞在 select 的 worker 立刻返回，先跑任务，再处理就绪 key。没有 wakeup，新连接要等到下一次网络事件才会被注册，可能白等很久。

注册失败时（比如 channel 已被对端关闭、selector 已关闭），任务里要关闭 channel，不能把描述符留给 GC。catch (IOException) 里再 close 一次，关闭失败就忽略。

ReactorGroup：把 boss 和 workers 组装起来

package com.plume.linkd.core;

import java.io.IOException;
import java.nio.channels.SocketChannel;
import java.util.concurrent.atomic.AtomicInteger;

public final class ReactorGroup implements AutoCloseable {
    private final EventLoop boss;
    private final EventLoop[] workers;
    private final AtomicInteger next = new AtomicInteger();

    public ReactorGroup(int workerCount) throws IOException {
        if (workerCount <= 0) throw new IllegalArgumentException("workerCount must be positive");
        this.boss = new EventLoop();
        this.workers = new EventLoop[workerCount];
        for (int i = 0; i < workerCount; i++) this.workers[i] = new EventLoop();
    }

    public void start(int port, HandlerFactory factory) throws IOException {
        for (int i = 0; i < workers.length; i++)
            new Thread(workers[i], "linkd-io-" + i).start();
        boss.listen(port, channel -> dispatch(channel, factory));
        new Thread(boss, "linkd-accept").start();
    }

    private void dispatch(SocketChannel channel, HandlerFactory factory) {
        EventLoop worker = workers[Math.floorMod(next.getAndIncrement(), workers.length)];
        worker.execute(() -> {
            try { worker.register(channel, factory.create(worker)); }
            catch (IOException e) {
                try { channel.close(); } catch (IOException ignored) { }
            }
        });
    }

    @Override public void close() {
        boss.shutdown();
        for (EventLoop worker : workers) worker.shutdown();
    }
}

四个点值得展开：

• 构造函数只建对象，不启动线程。 EventLoop 的构造只 Selector.open()，没有线程副作用；启动由 start 统一做，便于测试和优雅关闭。
• 线程命名固定。 linkd-io-N 和 linkd-accept 会出现在 jcmd Thread.print、JFR 和采样 profiler 里；排障时能一眼看出是 accept 线程还是 I/O 线程。
• round-robin 用 Math.floorMod。 AtomicInteger.getAndIncrement() 溢出后会变成负数，直接 % workers.length 会得到负下标，抛 ArrayIndexOutOfBoundsException；floorMod 把结果落在非负区间。
• 连接不迁移。 分配到哪个 worker 就一直在那里；迁移要把 SelectionKey、Connection、写队列整体搬过去，代价高且容易破坏线程归属。需要重新平衡时，宁可断开重连。

close() 只发停止信号，不等待线程退出。EventLoop.shutdown() 把 running 置 false 并 wakeup，run() 退出循环后在 finally 里关闭 selector。生产环境要 join 线程或等一个宽限期，14.3 会展开。

锁与可见性：只共享三样东西

多线程 Reactor 的热路径上没有锁。线程之间只共享三样东西：

除此之外，Connection、SelectionKey、Handler 的字段、outbound 队列都只属于一个 worker 线程，不需要 volatile，也不需要 synchronized。这是「线程封闭」而不是「靠锁保护」：只要没有别的线程拿到引用，就不存在竞争。

几条容易踩的边界：

• 不要在 boss 线程调用 connection.enqueue。 即使 Connection 对象能拿到，outbound 和 interestOps 也不是线程安全的；要走 worker.execute。
• 不要共享 handler 实例。 ReactorEchoHandler 里的 inbound 是复用缓冲区；如果同一个实例被两个 worker 使用，两个线程会同时 clear()、read()、flip()，数据会串。HandlerFactory 按连接创建，就是为了避开这一点。
• 不要用 synchronized 包住 register 来「修」线程问题。 锁不能让 Selector 变成线程安全，只会把 boss 和 worker 串在一起，增加死锁风险。正确做法是投递到目标线程。
• 指标计数器用 LongAdder。 如果要在多个 worker 上累计连接数、字节数，普通 long++ 会丢更新；LongAdder 比 AtomicLong 更适合高并发累加，读取时接受近似值。
• 跨线程关闭连接也要投递。 业务线程想主动断开某条连接，调用 worker.execute(connection::close)，不要直接 close()。

volatile 只提供可见性，不提供复合操作的原子性。running 是布尔开关，读写各一次，够用；next 需要「读旧值再加一」，所以用 AtomicInteger，不是 volatile int。

对照实验：同一脚本，1 个与 4 个 I/O 线程

实验用同一个 DelayEchoHandler，在 onReadable 里 Thread.sleep(5) 模拟一次阻塞业务处理，然后回显。客户端 8 个线程，每个发 25 次请求，共 200 次请求；服务端分别用 1 个和 4 个 I/O 线程。

package com.plume.linkd.core;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

public final class DelayEchoHandler implements Handler {
    private final long delayMillis;
    private final ByteBuffer inbound = ByteBuffer.allocate(64);

    public DelayEchoHandler(long delayMillis) { this.delayMillis = delayMillis; }

    @Override public void onReadable(Connection connection) {
        SocketChannel channel = connection.channel();
        try {
            inbound.clear();
            int n = channel.read(inbound);
            if (n == -1) { connection.close(); return; }
            if (n == 0) return;
            inbound.flip();
            Thread.sleep(delayMillis);
            ByteBuffer out = ByteBuffer.allocate(n);
            out.put(inbound).flip();
            connection.enqueue(out);
        } catch (IOException e) { connection.close(); }
        catch (InterruptedException e) { Thread.currentThread().interrupt(); connection.close(); }
    }

    @Override public void onWritable(Connection connection) { }
    @Override public void onClosed(Connection connection) { }
}

客户端每个线程的请求循环是「写一条、读满同样长度」：

for (int r = 0; r < requests; r++) {
    out.write(request); out.flush();
    int read = 0;
    while (read < buffer.length) {
        int n = in.read(buffer, read, buffer.length - read);
        if (n < 0) throw new EOFException("server closed");
        read += n;
    }
}

EchoBenchmarkServer 只是把 6.2 启动类的监听参数换成 worker 数、端口和延迟；EchoBenchmarkClient 用固定线程池起 8 个上面的循环。启动和测量：

java -cp out com.plume.linkd.app.EchoBenchmarkServer 1 9091 5 &
java -cp out com.plume.linkd.app.EchoBenchmarkClient 9091 8 25 single
java -cp out com.plume.linkd.app.EchoBenchmarkServer 4 9092 5 &
java -cp out com.plume.linkd.app.EchoBenchmarkClient 9092 8 25 multi4

本机 3 次运行（单次样本，未做统计）：

single clients=8 requestsPerClient=25 total=200 elapsedMs=1348
single clients=8 requestsPerClient=25 total=200 elapsedMs=1384
single clients=8 requestsPerClient=25 total=200 elapsedMs=1326
multi4 clients=8 requestsPerClient=25 total=200 elapsedMs=345
multi4 clients=8 requestsPerClient=25 total=200 elapsedMs=349
multi4 clients=8 requestsPerClient=25 total=200 elapsedMs=338

单线程 1326–1384 毫秒，4 个 I/O 线程 338–349 毫秒，约 3.9 倍。这个数字能对上算术：单线程要串行处理 200 次 5 毫秒睡眠，理论下限 1000 毫秒；4 个 worker 按 round-robin 各拿 2 个客户端，每个 worker 串行 50 次，理论下限 250 毫秒。剩下的几百毫秒和近百毫秒是客户端线程、系统调用、调度和输出开销。

三点限制必须说清：

• 这是睡眠，不是 CPU 计算。 Thread.sleep 让出 CPU，多线程的收益来自「一个线程睡着时另一个线程能跑」。如果 handler 是 CPU 密集的，4 个 I/O 线程最多用上 4 个核，收益取决于核数和任务性质，不能照搬这个倍数。
• 客户端只有 8 个。 每个 worker 分到 2 条连接，4 个 worker 已经能吃饱；继续加到 8 个 worker 时每个 worker 只分到 1 条连接，理论延迟还能下降，但客户端线程和调度开销会占更大比例，收益要重新测，不能从这个 3.9 倍外推。
• 单次运行，不当作吞吐上限。 这组数字证明的是「阻塞处理被分散到多个 I/O 线程」，不是「linkd 只能跑这个 QPS」。压测口径、连接复用、消息大小都会改变结果，15.2 会讲。

如果 handler 里是数据库查询、文件读写这类阻塞调用，多 I/O 线程仍然只是把阻塞分散开；真正的隔离是把业务处理丢进独立线程池，12.2 会讲。多 Reactor 解决的是 I/O 事件分发的并行度，不是业务阻塞。

常见错误与边界

错误一：boss 线程直接 register。 本机会抛 IllegalStateException: register must run on the event loop thread。必须 worker.execute，让注册发生在目标 selector 的线程。

错误二：多线程提交仍用 ArrayDeque。 两个线程同时 add 会破坏队列结构，任务可能丢失或抛 ConcurrentModificationException。换成 ConcurrentLinkedQueue，并保留 wakeup()。

错误三：多个 worker 共享一个 handler。 handler 里的复用缓冲区会被并发读写，回显数据串包。用 HandlerFactory 每个连接建一个，或者把 handler 做成无状态。

错误四：round-robin 用 % 而不是 floorMod。 AtomicInteger 溢出成负数后，% 会给出负下标，直接数组越界。Math.floorMod 是这类取模的正确写法。

错误五：accept 线程里做业务。 boss 线程一旦读数据、查数据库，accept 就停了，新连接堆在内核 backlog 里。boss 只做 accept 和分发。

错误六：从业务线程直接 connection.close()。 close 会改 SelectionKey、清空 outbound，这些状态属于 worker 线程。改成 worker.execute(connection::close)。

边界：worker 数量不是越多越好。 每个 worker 一个 selector、一个线程、一个 kqueue；超过核数后上下文切换和缓存失效会增加。从核数起步，按压测调整，不要按连接数开线程。

边界：多 Reactor 不解决阻塞 handler。 一个 worker 上的慢 handler 仍然会拖住该 worker 上的其他连接，只是影响范围从「全部连接」缩小到「1/N」。要彻底隔离，需要业务线程池或虚拟线程，12.2 会展开。

边界：连接迁移很难。 把一条已注册的连接从一个 selector 移到另一个，要处理未读完的半包、未写完的队列、SelectionKey 的取消和重新注册；本卷不做在线迁移，连接生命周期内固定在一个 worker。

小结

• 主从模型里 boss 只 accept 和分发，worker 各自持有 selector，只做已建立连接的读写。
• 跨线程注册必须 execute 到目标循环；直接 register 会抛 IllegalStateException。
• 任务队列要从 ArrayDeque 换成 ConcurrentLinkedQueue，wakeup() 让 worker 及时处理新任务。
• round-robin 用 AtomicInteger 加 Math.floorMod；连接固定在一个 worker，不迁移。
• 热路径没有锁：running 用 volatile，任务队列用并发队列，计数器用 AtomicInteger/LongAdder；连接状态靠线程封闭。
• 本机 8 客户端 × 25 请求、handler sleep 5 毫秒：单线程 1326–1384 毫秒，4 个 I/O 线程 338–349 毫秒；这是阻塞分散的效果，不是吞吐上限。

多线程 Reactor 把「等连接」和「读写连接」分开了，但 6.2 的 Echo 仍然假设一次 read 能拿到完整消息。下一节回到非阻塞读写本身：accept、read、write 各自可能只完成一部分，半包和粘包要在缓冲区里显式处理。

阅读导航：上一节：6.2 单线程 Reactor 实现 · 下一节：7.1 非阻塞 accept、read 与 write 。