本节目标：把 Selector 的注册、就绪、读写和关闭拼成一个可运行的单线程 Echo 服务，并说清每条连接的状态归谁所有。
适用版本：JDK 25（Temurin 25.0.5+8）；Selector、SocketChannel、ServerSocketChannel。
承接的问题
6.1 的探针只处理了一条连接的一轮读写。真实服务要同时接住任意多条连接，每条连接可能读一半、写一半、随时断开；把这些动作放进一个循环，就是单线程 Reactor。
这一节落地 BRIEF 7.2 的四个共享类型：BufferPool、Handler、Connection、EventLoop，包名 com.plume.linkd.core；启动类放在 com.plume.linkd.app，全部只用 JDK 25 标准库。
本机口径：Apple M1 Pro，10 核，32 GB，macOS 26.3 (25D125)；Temurin 25.0.5+8。下面所有运行输出都来自这台机器。
线程归属：一条连接只属于一个循环
单线程 Reactor 的核心约束不是「只有一个线程」，而是「每条连接的可变状态只由一个线程碰」。accept、read、write、close、interestOps 的修改，全部发生在同一个事件循环线程上：
linkd-io-0
Selector
├─ ServerSocketChannel OP_ACCEPT
├─ Connection A OP_READ | OP_WRITE
└─ Connection B OP_READ
tasks: ArrayDeque<Runnable>
三个「队列」要分清：accept backlog 是内核里的已完成连接队列，server.accept() 从里面取；selector 就绪集合是 selectedKeys()；每条连接的写队列是 Connection.outbound。没有单独的「读队列」：读事件就绪后 handler 直接从 socket 接收缓冲区读，读到多少算多少，剩下等下一次 OP_READ。这也是非阻塞读必须处理半包的原因，7.2 会展开。
因为只有一个线程，Connection.outbound、SelectionKey 的 interestOps、handler 的字段都不需要锁，串行化本身就是互斥。代价是 handler 里任何阻塞操作都会卡住所有连接：一次 5 毫秒的 Thread.sleep，在 1000 条连接上就是 5 秒的排队，6.3 会用实验量出这个差距。
Handler.onClosed 通过 Connection 的 close hook 触发，使每条连接只回调一次；无论是 handler 读到 -1 主动关闭，还是事件循环捕获 IOException 后关闭，最终都走同一个 close()。
共享类型：BufferPool 与 Handler
BufferPool 管直接缓冲区的复用。直接缓冲区分配和回收都比堆缓冲区贵，频繁 allocateDirect 会把压力推给 Cleaner；池化把常用大小的缓冲区留在进程内。
package com.plume.linkd.core;
import java.nio.ByteBuffer;
import java.util.concurrent.ConcurrentLinkedQueue;
public final class BufferPool {
private static final int MAX_CACHED = 64;
private final ConcurrentLinkedQueue<ByteBuffer> free = new ConcurrentLinkedQueue<>();
public ByteBuffer acquire(int minCapacity) {
ByteBuffer buffer;
while ((buffer = free.poll()) != null) {
if (buffer.capacity() >= minCapacity) { buffer.clear(); return buffer; }
}
return ByteBuffer.allocateDirect(Math.max(1024, minCapacity));
}
public void release(ByteBuffer buffer) {
if (buffer == null || !buffer.isDirect()) return;
buffer.clear();
if (free.size() < MAX_CACHED) free.offer(buffer);
}
}
5.1 和 5.2 的 BufferPool 用 ThreadLocal 绑线程，构造参数是 bufferSize 和 maxPerThread；6.2 为了把池作为 EventLoop 的普通字段、并在 6.3 多个 worker 之间复用同一实现，改成无构造参数的 ConcurrentLinkedQueue 版本，acquire/release 签名和 5.2 的「谁借谁还」所有权规则不变。acquire 返回的缓冲区已经 clear()：position 为 0，limit 等于 capacity，调用方写完要自己 flip()。release 只回收直接缓冲区，堆缓冲区交给 GC；free.size() 在 ConcurrentLinkedQueue 上是遍历计数，这里只做粗略上限。单线程版本用 ArrayDeque 就够，用并发队列是为了 6.3 的多个 I/O 线程能复用同一个类型；池上限按数量而不是字节数，大缓冲区场景要换成按字节配额，13.2 会讲。
Handler 只有三个回调：
package com.plume.linkd.core;
public interface Handler {
void onReadable(Connection connection);
void onWritable(Connection connection);
void onClosed(Connection connection);
}
接口不抛 IOException，因为事件循环无法替具体协议决定「读失败后是重试还是关闭」；handler 自己捕获异常并调用 connection.close()。onWritable 在 Connection.flush() 之后调用，给应用层一个「队列已排空」的通知，Echo 不需要，留空。
Handler 实例按连接创建。字段里的 inbound 因此只属于一条连接；即使同一个 handler 被多条连接复用，只要回调都在同一个线程串行执行，字段也不会并发访问。跨线程共享 handler 则是另一回事，6.3 会避免。
Connection：写队列与 interestOps
Connection 把通道、key、缓冲池和写队列绑在一起；应用只通过它入队和关闭，不直接碰 SelectionKey。
package com.plume.linkd.core;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.*;
public final class Connection {
private final SocketChannel channel;
private final SelectionKey key;
private final BufferPool pool;
private final Queue<ByteBuffer> outbound = new ArrayDeque<>();
private boolean closed;
private Runnable closeHook = () -> { };
Connection(SocketChannel channel, SelectionKey key, BufferPool pool) {
this.channel = channel; this.key = key; this.pool = pool;
}
public SocketChannel channel() { return channel; }
public void enqueue(ByteBuffer out) {
if (out == null) return;
if (closed || !key.isValid() || !out.hasRemaining()) { pool.release(out); return; }
outbound.add(out);
try { key.interestOps(key.interestOps() | SelectionKey.OP_WRITE); }
catch (CancelledKeyException e) { close(); }
}
boolean flush() throws IOException {
while (!outbound.isEmpty()) {
ByteBuffer buffer = outbound.peek();
channel.write(buffer);
if (buffer.hasRemaining()) return false;
outbound.poll();
pool.release(buffer);
}
if (key.isValid()) key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
return true;
}
void onClose(Runnable hook) { this.closeHook = hook; }
public void close() {
if (closed) return;
closed = true;
key.cancel();
try { channel.close(); } catch (IOException ignored) { }
ByteBuffer buffer;
while ((buffer = outbound.poll()) != null) pool.release(buffer);
closeHook.run();
}
}
四个方法各自只做一件事：
enqueue校验连接和缓冲区，入队后打开OP_WRITE；空缓冲区直接还给池子。flush从队头开始写，写完一个释放一个；部分写保留当前缓冲区并返回 false，队列排空后关掉OP_WRITE。close幂等：取消 key、关闭通道、释放所有未写出的缓冲区、执行 close hook。onClose由EventLoop.register注入，用来回调Handler.onClosed。
这里没有加锁，也不能加锁：outbound、closed、key.interestOps 都假设只在事件循环线程访问。业务线程直接调用 enqueue，ArrayDeque 可能损坏，interestOps 也可能和 select 竞争；正确方式是 eventLoop.execute(() -> connection.enqueue(buffer))。
close() 里的 key.cancel() 延迟生效，但 channel.close() 会立刻释放描述符。已排队的写缓冲区不会自动发出去，关闭前要么先排空，要么明确接受丢弃；14.3 会把「排空再关」做成可配置的优雅关闭。
EventLoop：accept、read、write 一次跑通
EventLoop 是这一节的主体。它持有 selector、缓冲池和任务队列，负责注册监听、接受连接、分发读写事件。
package com.plume.linkd.core;
import java.io.IOException;
import java.net.*;
import java.nio.channels.*;
import java.util.*;
public final class EventLoop implements Runnable {
private final Selector selector;
private final BufferPool pool = new BufferPool();
private final Queue<Runnable> tasks = new ArrayDeque<>();
private volatile boolean running = true;
private Thread loopThread;
public EventLoop() throws IOException { this.selector = Selector.open(); }
@FunctionalInterface
public interface ChannelAcceptor { void accept(SocketChannel channel) throws IOException; }
public void listen(int port, ChannelAcceptor acceptor) throws IOException {
ServerSocketChannel server = ServerSocketChannel.open();
server.setOption(StandardSocketOptions.SO_REUSEADDR, true);
server.bind(new InetSocketAddress("127.0.0.1", port), 128);
server.configureBlocking(false);
SelectionKey key = server.register(selector, SelectionKey.OP_ACCEPT);
key.attach(new AcceptorAttachment(acceptor));
}
public void register(SocketChannel channel, Handler handler) throws IOException {
if (loopThread != null && Thread.currentThread() != loopThread)
throw new IllegalStateException("register must run on the event loop thread");
channel.configureBlocking(false);
SelectionKey key = channel.register(selector, SelectionKey.OP_READ);
Connection connection = new Connection(channel, key, pool);
connection.onClose(() -> handler.onClosed(connection));
key.attach(new Attachment(connection, handler));
}
public void execute(Runnable task) { tasks.add(task); selector.wakeup(); }
public BufferPool buffers() { return pool; }
@Override public void run() {
loopThread = Thread.currentThread();
try {
while (running) { selector.select(); runTasks(); processSelectedKeys(); }
} catch (IOException e) {
if (running) e.printStackTrace();
} finally {
try { selector.close(); } catch (IOException ignored) { }
}
}
public void shutdown() { running = false; selector.wakeup(); }
private void runTasks() {
Runnable task;
while ((task = tasks.poll()) != null) {
try { task.run(); } catch (RuntimeException e) { e.printStackTrace(); }
}
}
private void processSelectedKeys() throws IOException {
Set<SelectionKey> selected = selector.selectedKeys();
Iterator<SelectionKey> iterator = selected.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
if (!key.isValid()) continue;
try {
if (key.channel() instanceof ServerSocketChannel server) {
try {
SocketChannel channel;
while ((channel = server.accept()) != null) {
try { ((AcceptorAttachment) key.attachment()).acceptor().accept(channel); }
catch (IOException e) { channel.close(); }
}
} catch (IOException e) {
if (running) e.printStackTrace();
}
continue;
}
Attachment attachment = (Attachment) key.attachment();
if (key.isReadable()) attachment.handler().onReadable(attachment.connection());
if (key.isValid() && key.isWritable()) {
attachment.connection().flush();
attachment.handler().onWritable(attachment.connection());
}
} catch (IOException | RuntimeException e) { closeKey(key); }
}
}
private void closeKey(SelectionKey key) {
if (key.attachment() instanceof Attachment current) current.connection().close();
key.cancel();
try { key.channel().close(); } catch (IOException ignored) { }
}
private record AcceptorAttachment(ChannelAcceptor acceptor) { }
private record Attachment(Connection connection, Handler handler) { }
}
几个设计点对着代码看：
listen只注册 accept。SO_REUSEADDR让重启不被 TIME_WAIT 挡住；backlog 128 是内核排队上限提示；绑定127.0.0.1是为了本机实验，对外服务要换0.0.0.0或指定网卡。register强制线程归属。loopThread在run()里赋值；别的线程直接调用会抛IllegalStateException: register must run on the event loop thread。单线程版的 accept 回调本来就在循环线程，这个检查不会触发；6.3 的跨线程分配会依赖它暴露错误。- 循环顺序是 select、任务、事件，读在写前。
execute用ArrayDeque加wakeup，单线程版只允许循环线程自己提交任务；先跑任务再处理就绪 key，handler 读完可能立刻enqueue并打开OP_WRITE，同一个 key 如果本次也可写，紧接着的flush会少一次select往返。accept 分支单独捕获IOException，描述符耗尽一类错误不会关掉监听 socket；连接 key 的异常才走closeKey，由它调用Connection.close()触发onClosed，再取消 key、关闭通道。
run() 里没有给 select() 设超时。停机靠 shutdown() 的 wakeup()；空闲检测、心跳和超时归 7.3 和 14.2，单线程版先把「事件驱动」跑通，不把定时逻辑混进来。
ReactorEchoHandler 与启动
Echo 的协议只有一句话：读到多少字节，原样写回多少字节。handler 负责读，Connection 负责排队和写。
package com.plume.linkd.core;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
public final class ReactorEchoHandler implements Handler {
private final BufferPool pool;
private final ByteBuffer inbound = ByteBuffer.allocate(4096);
public ReactorEchoHandler(BufferPool pool) { this.pool = pool; }
@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();
ByteBuffer out = pool.acquire(n);
out.put(inbound).flip();
connection.enqueue(out);
System.out.println("echo " + n + " bytes");
} catch (IOException e) { connection.close(); }
}
@Override public void onWritable(Connection connection) { }
@Override public void onClosed(Connection connection) {
System.out.println("closed " + connection.channel());
}
}
关键在 ByteBuffer out = pool.acquire(n); out.put(inbound).flip(); 这一行。不能把 inbound 直接 enqueue：它是 handler 的复用缓冲区，下一次 onReadable 会 clear() 并覆盖它，而写队列里的数据可能还没发完。必须把本次读到的 n 个字节复制到独立的出站缓冲区。
read 返回 -1 才是对端关闭，返回 0 只表示这次没有数据。onWritable 留空，因为真正的写出在 Connection.flush() 里完成；System.out.println 只是教学日志，生产代码应换成 System.getLogger。
启动类把监听、handler 工厂和关停钩子连起来：
public static void main(String[] args) throws Exception {
EventLoop loop = new EventLoop();
loop.listen(9090, channel ->
loop.register(channel, new ReactorEchoHandler(loop.buffers())));
Runtime.getRuntime().addShutdownHook(new Thread(loop::shutdown));
loop.run();
}
listen 的 lambda 在事件循环线程执行，所以可以直接 register。shutdown 把 running 置 false 并 wakeup，run 退出循环后在 finally 里关闭 selector，注册的通道随之关闭。
运行与行为验证
把四个类放进 linkd-core/src/main/java/com/plume/linkd/core/，启动类放进 linkd-app/src/main/java/com/plume/linkd/app/，编译运行：
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.app.ReactorMain
另开一个终端，用 Python 发一条消息并半关闭写方向：
import socket
s = socket.create_connection(("127.0.0.1", 9090), timeout=3)
s.sendall(b"hello\n")
print("client recv=" + s.recv(64).decode().strip())
s.shutdown(socket.SHUT_WR)
print("client eof=" + str(s.recv(1) == b""))
s.close()
本机输出（客户端）：
client recv=hello
client eof=True
本机输出（服务端）；客户端收到完整 6 字节并读到 EOF，服务端随后触发 onClosed：
echo 6 bytes
closed java.nio.channels.SocketChannel[closed]
这条链路里没有线程池、没有第三方库，只有 selector、通道和两个缓冲区。
常见错误与边界
错误一：在 handler 里做阻塞操作。 单线程 Reactor 的所有连接共享一个线程。handler 里一次 5 毫秒的数据库查询，会让排队在它后面的所有连接一起等 5 毫秒。业务阻塞要移到业务线程池，12.2 会讲隔离方式。
错误二：从业务线程直接 connection.enqueue。 outbound 是 ArrayDeque，不是线程安全的；interestOps 也会和 select 竞争。正确写法是 loop.execute(() -> connection.enqueue(buffer))，让入队发生在本线程。
错误三：跨线程直接 register。 本机运行会抛 java.lang.IllegalStateException: register must run on the event loop thread。多线程版必须先 execute 到目标循环，6.3 会给出代码。
错误四：把 inbound 直接入队。 下一次读会覆盖它，客户端可能收到重复或错乱的数据。出站数据必须复制到独立缓冲区，或者每次读都新分配一个。
错误五：把 read 返回 0 当成连接关闭。 非阻塞读返回 0 只是「这次没有数据」，继续等下一次 OP_READ；只有 -1 才是 EOF。把 0 当 EOF 会频繁误关连接。
错误六：写队列无上限。 客户端读得慢，outbound 会一直增长，最后 OOM。需要按连接设置高水位，超过就停止读或断开，14.1 会讲背压。
边界：单线程 Reactor 用不满多核。 它解决的是「大量空闲连接」的线程开销，不是「CPU 密集」的并行；CPU 密集型 handler 要交给业务线程池，多 I/O 线程只能分散阻塞的 I/O 处理。虚拟线程也不是这个循环的替代品：它适合「一个请求一个线程」的阻塞式业务代码，Selector 循环仍然是平台线程上的事件分发器，两者可以组合，但不能把 select() 放进虚拟线程就指望它变成多路复用。
小结
- 单线程 Reactor 用一条线程、一个
Selector管理多条连接；accept、read、write、close 全部串行。 BufferPool复用直接缓冲区；acquire返回已清空的缓冲区，写完后要flip，关闭时逐个release。Connection用outbound队列和OP_WRITE表达背压；部分写保留缓冲区，排空后关闭写事件。EventLoop的顺序是select、跑任务、处理就绪 key；accept 异常不关闭监听 socket，连接异常才走closeKey。handler 在事件循环线程执行，不能阻塞；跨线程只能execute，不能直接改Connection或register，出站数据必须复制到独立缓冲区。
单线程版能接住大量空闲连接，但一个慢 handler 会拖住所有连接。下一节把 accept 和 I/O 拆到不同线程：一个 accept 线程负责收连接，多个 I/O 线程各自持有 selector，用 round-robin 分配，并用同一套压测脚本量出差距。
阅读导航：上一节：6.1 Selector 就绪事件与 interestOps · 下一节：6.3 多线程 Reactor 与主从模型 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。