03 IO 模型与高并发

本篇定位：从五种 IO 模型的定义出发，穿过 select/poll/epoll 的内核实现，落地到 Java 的 BIO/NIO/AIO、Reactor 模式与 Netty 线程模型，最后给出工业采集场景特有的并发模型——为什么"每设备一线程"在车间里往往合理、为什么一条 PLC 连接上必须串行化、事务 ID 流水线化的可能与风险，以及一份可直接改造使用的采集调度器骨架代码。TCP 传输层细节（窗口、重传、Nagle）见 01-TCP协议深入.md，报文边界与拆帧见 02-粘包与拆包.md。

本篇目录

• 1. 五种 IO 模型：两阶段阻塞对比
• 2. Linux 底层：select / poll / epoll
• 3. Java BIO：阻塞模型与完整多线程服务端
• 4. Java NIO：Buffer / Channel / Selector
• 5. Java AIO：为什么 Linux 上不如 NIO
• 6. Reactor 模式：三种形态
• 7. Netty 线程模型与 ByteBuf
• 8. 零拷贝：sendfile / mmap / transferTo
• 9. 线程池设计：队列、线程数、拒绝策略、虚拟线程
• 10. 工业采集的真实并发模型
• 11. 性能测试方法
• 小结
• 常见误区

1. 五种 IO 模型：两阶段阻塞对比

一次 recv() 在内核里其实是两个阶段：阶段一等待数据（报文到达并拷入内核缓冲区），阶段二拷贝数据（内核缓冲区 → 用户空间）。POSIX 定义：只要阶段二阻塞了进程就是同步 IO；两阶段都不阻塞才是异步 IO。

  应用进程                     内核
     |  recv()                   |  ← 阶段一：等待数据准备好
     |-------------------------> |    ...报文到达，DMA 拷入内核缓冲...
     |  <----------------------  |  ← 阶段二：内核缓冲 → 用户空间拷贝

  阻塞IO     ====阻塞 recv=====================> 拷贝完成返回
                [等待数据.....][拷贝]
  非阻塞IO   recv=EWOULDBLOCK  recv==> 拷贝完成
                [轮询CPU空转]  [拷贝]
  IO多路复用 ====阻塞 select(多fd)=========> 就绪 ==> recv 拷贝
                [等任一fd就绪...]          [拷贝]
  信号驱动   （继续执行）...收到SIGIO...====> recv 拷贝
  异步IO     （继续执行）........收到"已完成"通知（内核完成等待+拷贝）
  ←──────────────────── 时间轴 ────────────────────→

关键结论：IO 多路复用是同步 IO，它高效的原因不是"不阻塞"，而是"一个线程能同时等很多个 fd"。

2. Linux 底层：select / poll / epoll

  select/poll 的 O(n) 问题：连接 n=10000 但每次只有 10 个活跃 → 每次调用仍要
  拷贝 10000 个 fd、内核遍历 10000 次、返回后用户态再遍历 10000 次，开销与 n 成正比。

  epoll 的做法：epoll_create() 建 eventpoll 对象
    ├─ rbr     ：红黑树，存所有被监视的 fd（epoll_ctl ADD/DEL/MOD，O(log n)）
    └─ rdllist ：双向链表，存已就绪的 fd（epoll_wait 直接返回，O(就绪数)）
    唤醒路径   ：网卡中断 → 协议栈 → 回调把就绪 fd 挂进 rdllist

  用户态                              内核 eventpoll
   epoll_ctl(ADD/MOD/DEL) ───────> ┌─────────┐  ┌──────────────┐
                                   │ 红黑树   │  │ 就绪链表      │
   epoll_wait() <────────────────  │(所有fd) │  │ rdllist(就绪) │
    返回就绪事件数组                └─────────┘  └──────────────┘
                                       ^              ^
                                    epoll_ctl    设备中断回调挂入

2.1 LT 与 ET

// ET 模式下的正确读法（NIO 层面对应结构）
while (true) {
    int n = channel.read(buf);
    if (n > 0) { handle(buf); buf.clear(); continue; }
    if (n == 0) break;          // 非阻塞且无数据 → 本次已读干净
    if (n < 0) { close(); break; }
}
// ⚠ ET 下只读一次就返回 → 剩余数据留在缓冲区且不再被通知

2.2 EPOLLONESHOT

EPOLLONESHOT 让 fd 在一次事件被处理后自动从监视中移除，必须显式 epoll_ctl(MOD) 重新武装。价值是防止多线程同时处理同一个 fd：

  无 ONESHOT：线程A 拿到 fd=7 开始读，线程B 同时也拿到 fd=7 → 两线程交错读
              → 数据错序、协议解析崩溃
  有 ONESHOT：fd=7 只投递给一个线程一次，处理完才重新注册 → 天然串行化

Netty 的 NioEventLoop 本身保证一个 Channel 只由一个线程处理（效果等价）；原生 epoll 传输下会用到 oneshot（需实测确认版本行为）。补充：EPOLLEXCLUSIVE（Linux 4.5+）用于缓解多 epoll 实例等待同一 fd 的惊群。

3. Java BIO：阻塞模型与完整多线程服务端

阻塞点分布：accept() 等新连接、read() 等数据、write() 等发送缓冲腾出空间、connect() 等握手 → "等待"永远占据一个线程，线程数 ≈ 并发连接数。

package com.example.iot.bio;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.concurrent.*;

/** 多线程 BIO 服务端：一个连接一个线程，读取后回显。 */
public final class BioEchoServer {

    private final int port;
    private final ThreadPoolExecutor pool;
    private volatile ServerSocket serverSocket;

    public BioEchoServer(int port, int maxThreads) {
        this.port = port;
        // 有界队列 + CallerRunsPolicy：连接突增时反压而不是堆积到 OOM
        this.pool = new ThreadPoolExecutor(8, maxThreads, 60L, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(256),
                r -> { Thread t = new Thread(r); t.setDaemon(true); return t; },
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    public void start() throws IOException {
        serverSocket = new ServerSocket(port, 512);          // backlog 显式指定
        System.out.println("BIO server listen on " + port);
        while (!serverSocket.isClosed()) {
            Socket conn;
            try {
                conn = serverSocket.accept();                 // 阻塞：等新连接
            } catch (IOException e) {
                if (serverSocket.isClosed()) break;
                System.err.println("accept 失败: " + e.getMessage());
                continue;
            }
            pool.execute(() -> handle(conn));                 // 每连接一线程
        }
    }

    private void handle(Socket conn) {
        try (conn) {
            conn.setTcpNoDelay(true);                         // 关 Nagle（见 01 篇）
            conn.setSoTimeout(10_000);                        // 读超时，防线程永久卡死
            InputStream in = conn.getInputStream();
            OutputStream out = conn.getOutputStream();
            byte[] buf = new byte[4096];
            while (true) {
                int n;
                try { n = in.read(buf); }                     // 阻塞：等数据
                catch (SocketTimeoutException e) { continue; } // 空闲超时，继续等
                if (n < 0) break;                             // 对端关闭
                out.write(buf, 0, n);
                out.flush();
            }
        } catch (IOException e) {
            System.err.println("连接处理异常: " + e.getMessage());
        }
    }

    public void stop() throws IOException {
        if (serverSocket != null) serverSocket.close();
        pool.shutdown();
    }

    public static void main(String[] args) throws Exception {
        BioEchoServer server = new BioEchoServer(9000, 200);
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try { server.stop(); } catch (IOException ignored) { } }));
        server.start();
    }
}

优点：代码直观，与"发→收→解析"的协议交互天然契合；超时/重试/断线重连逻辑最好写；异常栈清晰，便于与抓包对照；与"一条连接必须串行"的 PLC 语义天然匹配。 缺点：线程数随连接数线性增长（每线程默认栈 1MB，可用 -Xss 调小）；连接多而活跃少时上下文切换浪费；不适合 C10K 级公网服务。

3.1 为什么工业采集场景下"每设备一线程"可以接受

关键在于量级和连接利用率与互联网服务完全不同：

  互联网服务：10万+ 连接，频繁创建销毁，长尾客户端 → 必须 NIO/Netty
  工业采集  ：几十 ~ 几百台设备（单采集器通常 < 500）；连接长期保持；
              每连接利用率极低（周期 500ms、单次交互 2ms → 占用率 0.4%）

  算账：200 台设备 × 1 线程 = 200 线程 → 内存 200 × 512KB（-Xss512k）= 100MB
        99.6% 时间在阻塞，CPU 开销可忽略；代价远小于"手写 NIO 状态机 + 拆包
        + 重连"带来的 bug 成本

判断阈值（需实测确认）：单采集器设备数 < 1000、且不需要同时接收大量设备主动上报连接时，BIO/每设备一线程是最优工程选择；超过该量级或要做"设备主动上报"的服务端时，上 Netty。

4. Java NIO：Buffer / Channel / Selector

Buffer 状态机（最容易写错）：

  写模式 → flip() → 读模式 → clear()/compact() → 写模式

  capacity=8，写入 3 字节：position=3, limit=8
  flip() 后：position=0, limit=3     ← limit 变成"有效数据长度"
  compact()：把未读部分拷到开头 → position=1, limit=8

  忘 flip() → 读到未初始化数据；忘 clear() → 缓冲区被写满

4.1 完整 NIO Echo Server

package com.example.iot.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Set;

/** 单 Reactor 单线程 NIO Echo Server：一个 Selector 管所有连接。 */
public final class NioEchoServer {
    public static void main(String[] args) throws IOException {
        Selector selector = Selector.open();
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.bind(new InetSocketAddress(9001));
        ssc.configureBlocking(false);
        ssc.register(selector, SelectionKey.OP_ACCEPT);
        ByteBuffer buf = ByteBuffer.allocate(4096);
        while (true) {
            if (selector.select(1000) == 0) continue;         // 阻塞等事件，最多 1s
            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> it = keys.iterator();
            while (it.hasNext()) {
                SelectionKey key = it.next();
                it.remove();                                  // 必须移除，否则重复处理
                try {
                    SocketChannel ch;
                    if (key.isAcceptable()) {
                        ch = ((ServerSocketChannel) key.channel()).accept();
                        ch.configureBlocking(false);
                        ch.setOption(StandardSocketOptions.TCP_NODELAY, true);
                        ch.register(selector, SelectionKey.OP_READ);
                    } else if (key.isReadable()) {
                        ch = (SocketChannel) key.channel();
                        buf.clear();
                        int n = ch.read(buf);                 // 一次读到的字节数不确定
                        if (n < 0) { key.cancel(); ch.close(); continue; }
                        if (n == 0) continue;
                        buf.flip();
                        byte[] data = new byte[buf.remaining()];
                        buf.get(data);
                        String echo = "echo:" + new String(data, StandardCharsets.UTF_8);
                        key.attach(ByteBuffer.wrap(echo.getBytes(StandardCharsets.UTF_8)));
                        key.interestOps(SelectionKey.OP_WRITE);          // 有数据要写才注册
                    } else if (key.isWritable()) {
                        ch = (SocketChannel) key.channel();
                        ByteBuffer out = (ByteBuffer) key.attachment();
                        ch.write(out);                                   // 可能未写完
                        if (!out.hasRemaining()) {                       // 写完立刻取消关注
                            key.attach(null);
                            key.interestOps(SelectionKey.OP_READ);
                        }
                    }
                } catch (IOException e) {
                    key.cancel();
                    try { key.channel().close(); } catch (IOException ignored) { }
                }
            }
        }
    }
}

4.2 两个必知陷阱

陷阱一：OP_WRITE 不要常驻注册。 写缓冲区几乎总是"可写"，常驻会让 select() 立刻返回、CPU 空转 100%。正确做法：有数据要写时注册，写完立刻取消。

陷阱二：Selector 空轮询 bug（JDK-6670302）。 某些情况下 select() 不阻塞地立即返回 0，while 循环空转导致 CPU 100%。对策：统计连续空 select 次数，超过阈值（Netty 用 512）就重建 Selector 并重新注册 Channel；或直接用 Netty（其 NioEventLoop 内置了 rebuild 逻辑）。官方称后续 JDK 已修复，但生产仍建议保留重建逻辑（需实测确认）。

4.3 为什么手写 NIO 很痛苦

半包/粘包（read() 字节数不确定，必须自己做累积缓冲与拆帧，详见 02-粘包与拆包.md）、写半包（write() 不保证写完，要维护待写队列并动态注册 OP_WRITE）、Buffer 状态（flip/clear/compact 用错一个就出诡异数据）、空轮询（需自己重建 Selector）、连接管理（超时/重连/心跳/优雅关闭全自己写）、线程模型（单 Selector 是瓶颈，多 Selector 要自己分组与负载均衡）、内存（每连接缓冲区与直接内存的分配回收）。

结论：学 NIO 是为了理解原理与读懂 Netty；生产代码不要手写 NIO。

5. Java AIO：为什么 Linux 上不如 NIO

AsynchronousSocketChannel ch = AsynchronousSocketChannel.open();
ByteBuffer buf = ByteBuffer.allocate(256);
ch.connect(new InetSocketAddress(host, 502), buf, new CompletionHandler<Void, ByteBuffer>() {
    @Override public void completed(Void r, ByteBuffer b) {          // 连接完成 → 发起读
        ch.read(b, b, new CompletionHandler<Integer, ByteBuffer>() {
            @Override public void completed(Integer len, ByteBuffer bb) { bb.flip(); /* 处理再读 */ }
            @Override public void failed(Throwable exc, ByteBuffer bb) { /* 失败处理 */ }
        });
    }
    @Override public void failed(Throwable exc, ByteBuffer b) { /* 失败处理 */ }
});

  1) Linux 原生 AIO（io_submit/io_getevents）历史上主要面向磁盘 IO，对 socket
     支持长期不完善（直到 io_uring 才真正改善）
  2) 因此 JDK 在 Linux 上的 AIO 是"线程池 + epoll 模拟"的：内部线程做 select，
     再把完成事件投递给 CompletionHandler → 比直接用 NIO 多一层线程与队列
  3) Windows 上 AIO 走 IOCP，是真异步 —— "AIO 更好"的说法多来自 Windows
     经验，搬到 Linux 就反了；回调式 API 处理"请求-响应配对 + 拆包 + 超时
     重传"时可读性也差（回调地狱）

6. Reactor 模式：三种形态

  ① 单 Reactor 单线程
    +----------------------------------------------+
    | 单线程：select → accept → read → 业务 → write  |
    +----------------------------------------------+
    优点：无锁、无上下文切换；缺点：业务一阻塞全卡死，用不了多核；典型：Redis

  ② 单 Reactor 多线程
     ┌───────────┐   业务    ┌──────────────┐
     │Reactor线程 │ ───────> │ Worker 线程池 │
     │select/IO  │          │   业务处理    │
     └───────────┘          └──────────────┘
    优点：业务不阻塞 IO 线程；缺点：Reactor 线程仍是单点

  ③ 主从 Reactor 多线程（Netty 采用）
     客户端 ─> ┌───────────┐  注册  ┌─────────────┬─────────────┐
              │MainReactor│ ─────> │SubReactor-1 │SubReactor-N │
              │  (Boss)   │        │  (Worker)   │  (Worker)   │
              └───────────┘        └──────┬──────┴──────┬──────┘
                                          v             v
                                   ┌────────────────────────────┐
                                   │ 业务线程池（可选，隔离耗时操作）│
                                   └────────────────────────────┘

线程数与适用性：单 Reactor 单线程（1 线程，接收设备上报可用但业务不能阻塞）；单 Reactor 多线程（1+N，够用）；主从 Reactor 多线程（1+N+M，Netty 默认形态，最通用）。

Doug Lea《Scalable IO in Java》的分层是理解 Netty 的最佳地图：

  自顶向下：业务处理层 → 事件分发层(Dispatcher) → 事件循环层(Reactor)
          → 多路复用层(select/epoll 封装) → 连接管理层(Acceptor)

  设计原则：IO 事件与业务处理解耦（线程池隔离）；一个连接的所有事件尽量在
           同一线程内处理（无锁串行化）；多 Reactor 分摊 accept 与 IO 压力

7. Netty 线程模型与 ByteBuf

7.1 EventLoopGroup / Boss / Worker

EventLoopGroup boss   = new NioEventLoopGroup(1);      // 只负责 accept
EventLoopGroup worker = new NioEventLoopGroup();       // 默认 2 × CPU 核数
try {
    ServerBootstrap b = new ServerBootstrap();
    b.group(boss, worker)
     .channel(NioServerSocketChannel.class)
     .option(ChannelOption.SO_BACKLOG, 1024)                // 全连接队列
     .childOption(ChannelOption.TCP_NODELAY, true)            // 关 Nagle！
     .childOption(ChannelOption.SO_KEEPALIVE, true)
     .childOption(ChannelOption.SO_RCVBUF, 64 * 1024)
     .childHandler(new ChannelInitializer<SocketChannel>() {
         @Override protected void initChannel(SocketChannel ch) {
             ch.pipeline().addLast(new ReadTimeoutHandler(5, TimeUnit.SECONDS));
             ch.pipeline().addLast(new ModbusFrameDecoder());      // 拆帧（见 02 篇）
             ch.pipeline().addLast(new ModbusRequestHandler());    // 业务
         }
     });
    b.bind(502).sync().channel().closeFuture().sync();
} finally { boss.shutdownGracefully(); worker.shutdownGracefully(); }

  Boss NioEventLoopGroup(1)
    └─ NioEventLoop-1 ── selector ── ServerSocketChannel(OP_ACCEPT)
  Worker NioEventLoopGroup(N)                （每个 loop 一条线程）
    ├─ NioEventLoop-1 ── selector ── Channel A, Channel D ...
    ├─ NioEventLoop-2 ── selector ── Channel B, Channel E ...
    └─ NioEventLoop-N ── selector ── Channel C, Channel F ...

  ⚠ 常见错误：在 ChannelHandler 里做阻塞操作（查库、同步 HTTP、Thread.sleep）
    → 会卡住该 EventLoop 上绑定的所有 Channel
    → 耗时业务必须丢到独立业务线程池：pipeline.addLast(业务线程池, handler)

串行无锁化：一个 Channel 一旦注册到某个 EventLoop 就终身绑定同一线程，其所有 inbound/outbound 事件、定时任务、用户提交的任务都串行执行 → handler 里无需加锁，也没有可见性问题。

7.2 ChannelPipeline

  入站(Inbound) ────────────────────────────────────────────>
  head ─> [ssl] ─> [frameDecoder] ─> [idleState] ─> [business] ─> tail
     <───────────────────────────────────────────────────────
  <────────────────── 出站(Outbound) ───────────────────

  · inbound（channelRead/channelActive/exceptionCaught）从 head 向 tail 传播
  · outbound（write/flush/connect/close）从 tail 向 head 传播
  · 双向链表，节点是 ChannelHandlerContext，可 addLast/addFirst/addBefore
  · ctx.write(msg) 从"当前节点"向前找最近 OutboundHandler（跳过不必要 handler）；
    channel.write(msg) 从 tail 开始，走完全部 outbound handler

  典型采集客户端 Pipeline：
    入站 ByteBuf 字节流 → LengthFieldBasedFrameDecoder(完整帧)
       → ModbusResponseDecoder(帧 → Response 对象) → 业务 Handler
    出站 Request 对象 → ModbusRequestEncoder(→ ByteBuf) → 网卡
    ByteToMessageDecoder 帮你累积半包，正是 02 篇"长度前缀拆帧"的标准落地

7.3 ByteBuf

  与 ByteBuffer 的核心差别：ByteBuffer 单个 position，读写靠 flip() 切换；
  ByteBuf 用 readerIndex/writerIndex 双指针、读写互不干扰，支持动态扩容
  （受 maxCapacity 限制）、引用计数与池化。

  引用计数规则（工业服务最容易泄漏内存的地方）：① 谁最后使用谁 release，Pipeline
  中最后一个 InboundHandler 必须释放；② 继承 SimpleChannelInboundHandler<T>
  → Netty 自动 release；③ 交给业务线程池异步处理时必须在业务线程 release
  （注意跨线程配对）；④ 泄漏检测 -Dio.netty.leakDetection.level=PARANOID
  （默认 SIMPLE），生产别长期开 PARANOID，性能损耗大

8. 零拷贝：sendfile / mmap / transferTo

8.1 传统 read + write

  ① read()  ：DMA 把磁盘数据拷到内核页缓存          [DMA 拷贝]
  ②         ：CPU 把内核页缓存拷到用户缓冲区        [CPU 拷贝]
  ③ write() ：CPU 把用户缓冲区拷到 socket 发送缓冲  [CPU 拷贝]
  ④         ：DMA 把 socket 缓冲拷到网卡            [DMA 拷贝]

  合计 4 次拷贝（2 CPU + 2 DMA）+ 4 次上下文切换（read/write 各进出内核态 2 次）

   用户空间 ┌──────────┐
            │用户缓冲区 │ ←② CPU→ ③ CPU
   内核空间 │页缓存     │ ←① DMA      │socket缓冲│ →④ DMA→ 网卡
            └──────────┘             └─────────┘

8.2 三种优化路径

// 1) transferTo：底层用 sendfile，适合"文件 → 网络"的零拷贝
try (FileChannel file = FileChannel.open(Paths.get("history.dat"), StandardOpenOption.READ);
     SocketChannel sock = SocketChannel.open(new InetSocketAddress(host, port))) {
    long pos = 0, size = file.size();
    while (pos < size) {
        long n = file.transferTo(pos, size - pos, sock);   // 返回实际传输字节数
        if (n <= 0) break;                                 // 必须循环，可能未传完
        pos += n;
    }
}
// 2) MappedByteBuffer：mmap，适合大文件随机读
try (FileChannel file = FileChannel.open(p, StandardOpenOption.READ)) {
    MappedByteBuffer map = file.map(FileChannel.MapMode.READ_ONLY, 0, file.size());
    byte b = map.get(1024);                                // 像访问数组一样访问文件
}
// ⚠ MappedByteBuffer 释放依赖 GC（JDK 9+ 可用 Unsafe.invokeCleaner，JDK 14+ 有
//   MemorySession/Arena 等新 API，需实测确认）；频繁映射小文件会造成虚拟内存与
//   句柄压力 → 不要每个请求 map 一次

Netty 语境下的"零拷贝"是另一层含义，别混淆：OS 层靠 FileRegion（底层 transferTo）；用户层靠 CompositeByteBuf、ByteBuf.slice()、Unpooled.wrappedBuffer()；以及池化 ByteBuf 减少分配与复制。

9. 线程池设计：队列、线程数、拒绝策略、虚拟线程

9.1 为什么不能用无界队列

ExecutorService bad = Executors.newFixedThreadPool(16);   // 内部是无界 LinkedBlockingQueue

  故障链（现场真实会发生）：交换机故障 → 200 台 PLC 全部超时 → 失败任务进入
  重试队列 → 无界队列持续膨胀（每个任务还持有请求 ByteBuf 与上下文）
  → 内存上涨 → Full GC → OOM；更糟的是队列里的任务即使已无人关心也会执行，
  网络恢复瞬间几万任务一起涌出，把刚恢复的 PLC 再次打死

  正确做法：① 有界队列 ArrayBlockingQueue（容量 = 线程数 × 单线程允许积压量）
  ② 队列满时执行拒绝策略或让提交方阻塞（反压）③ 任务入队记 enqueueTime，
  出队检查是否过期 → 直接丢弃 ④ 按设备维度限流：单设备积压超 N 个就丢最旧的

9.2 线程数估算

  CPU 密集：N_threads ≈ N_cpu（或 N_cpu + 1）
  IO 密集（通用公式）：N_threads ≈ N_cpu × (1 + 等待时间/计算时间)
  IO 密集（工程经验）：N_threads ≈ N_cpu × 2 ~ N_cpu × 4，再压测调整

  但工业采集是"设备数驱动"而非"CPU 驱动"：
    上限由"可并行的设备数"决定。例：200 台设备 → 并行度 32~64（跨设备并行），
    单设备内部串行 → 总线程数可控
    判据：队列常非空 + CPU 没满 → 加大并行度；
         CPU 满 + 队列空 → 瓶颈在计算（解析/序列化/落库），别再加线程

9.3 拒绝策略

9.4 虚拟线程（Loom，JDK 21 正式，JEP 444）

  机制：由 JVM 调度，挂载在少量 carrier 线程上；阻塞时自动卸载（yield），不占
  carrier → 创建百万个也不吃内存。价值：可以用"同步阻塞的 BIO 写法"获得接近
  NIO 的并发能力，Executors.newVirtualThreadPerTaskExecutor() 一行就够；
  "每设备一线程"从"勉强可接受"变成"几乎免费"，且同步代码 + 超时/重试逻辑的
  可读性远好于 Netty 回调链。

  限制与坑：① synchronized 代码块/方法会 pin（钉住）carrier 线程 → 阻塞时不
  卸载，退化成平台线程 → 改用 ReentrantLock（JDK 24 JEP 491 改进了 pinning，
  需实测确认）；② 不适合 CPU 密集任务（只提供并发不提供并行）；③ ThreadLocal
  滥用会放大对象数量（可用 ScopedValue 替代，JDK 21 为预览）；④ 监控指标含义
  变了（没有"活跃线程数"概念）；⑤ 不要拿虚拟线程池去跑 Netty 的 EventLoop

10. 工业采集的真实并发模型

10.1 并发需求怎么算

  真实并发需求 ≠ 设备数，而是"单位时间内的请求数"：
    请求数/秒 = Σ( 该设备测点数 / 单次请求能读的测点数 ) / 采集周期(秒)

  例：200 台设备 × 50 测点，周期 1s
    · 一个测点一次请求（最笨）：200 × 50 = 10000 req/s ← 做不到也没必要
    · Modbus 单次读保持寄存器最多 125 个（0x7D）→ 50 测点 1 次读完 → 200 req/s
      （连续地址还能再合并）；S7 一次 PDU 可读几百字节（取决于协商出的 PDU
      size，常见 240/480/960，需实测确认）→ 同理批量

  结论：第一步是"批量合并读"，把请求数降一个数量级，而不是先优化 IO 模型。
       批量合并是数量级收益，IO 模型优化通常只有百分比收益。

10.2 连接池 vs 每设备一连接

PLC 的连接数是硬约束，这是与互联网服务最大的差别：

  硬约束带来的设计结论：
  1) 连接是稀缺资源 → 必须全局记账：
     已用 = 采集器数 × 每采集器连接数 + HMI + 上位机 + 调试 PG
     超限 → PLC 直接拒绝新连接（connect 失败或握手后立刻 RST）
  2) "每设备一连接"优先于"连接池"：PLC 不像数据库，建连代价不只是 TCP 握手，
     还有 S7 的 COTP 连接协商与参数交换（协商 PDU size 等），代价远高 →
     连接应长期持有、断线按退避重连，而不是用完归还。连接池只在"同设备多种
     访问模式"时才有意义（如读写分离、优先级分离）
  3) 多采集器访问同一 PLC：明确分配连接预算（采集器 3、HMI 1、预留 4）；
     采集器要能感知"连接被拒"并降级（降低频率而不是疯狂重连）
  4) 断线重连必须带退避 + 抖动，否则整车间同时重连 = 连接风暴

10.3 一个连接上的 IO 必须串行化

  错误做法（多线程并发写同一 socket）：
    线程A: write(读DB1) ─┐
    线程B: write(读DB2) ─┴→ 线路上字节交错 |A前半|B|A后半| → PLC 解析出乱码

  即使写入是原子的，还有响应错配问题：A 发"读温度"、B 发"读压力"，回来两个
  响应无法区分归属 → 数据错位。⚠ 这是最危险的一类 bug：数据看起来正常但值是错的

  正确做法：一条连接 = 一个串行队列，同一时刻只有一个请求在飞（见 10.5）

10.4 用事务 ID 做流水线的可能与风险

响应带回相同标识 → 可以并发多个在飞请求（流水线），吞吐可提升数倍，尤其在高 RTT 链路：

  串行（RTT=2ms）：|--req1-->|--resp1--|--req2-->|--resp2--|  吞吐 = 1/RTT = 500 req/s
  流水线（窗口3） ：|--req1-->|--req2-->|--req3-->|          吞吐 ≈ 3/RTT = 1500 req/s
                   |<--resp1-|<--resp2-|<--resp3-|

但工业场景风险很大，必须逐条确认：

实践建议：默认串行；只有 ① 手册明确说明支持、② 实测验证过、③ 窗口限制 2~4、④ 只对读操作开放、⑤ 可用开关一键关闭，才启用流水线。

10.5 采集调度器骨架代码

设计要点：跨设备并行、单设备串行、优先级排队、任务超时、故障标记。线程数不随设备数增长（设备 500、并行度 32 也只需 32 个线程）。

package com.example.iot.sched;

import java.util.Comparator;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

/** 采集调度器：跨设备并行、单设备串行、优先级排队、任务超时。 */
public final class AcquisitionScheduler implements AutoCloseable {

    /** 在一条设备连接上执行的一次交互（读/写）。实现类内部必须保证线程安全。 */
    @FunctionalInterface
    public interface Command<T> {
        T execute(DeviceConnection conn) throws Exception;
    }

    /** 设备连接占位：真实实现持有 Socket/SocketChannel + 编解码器 + 重连逻辑。 */
    public static final class DeviceConnection {
        public final String deviceId;
        volatile boolean healthy = true;           // 置 false 后由重连线程处理
        DeviceConnection(String deviceId) { this.deviceId = deviceId; }
    }

    private static final class Task {
        final String deviceId;
        final int priority;                        // 数字越小越优先
        final long timeoutMs;
        final Command<?> command;
        final long enqueueTime = System.currentTimeMillis();
        final CompletableFuture<Object> future = new CompletableFuture<>();
        Task(String deviceId, int priority, long timeoutMs, Command<?> command) {
            this.deviceId = deviceId; this.priority = priority;
            this.timeoutMs = timeoutMs; this.command = command;
        }
    }

    private static final class DeviceQueue {
        final PriorityBlockingQueue<Task> queue = new PriorityBlockingQueue<>(256,
                Comparator.comparingInt((Task t) -> t.priority)
                          .thenComparingLong(t -> t.enqueueTime));
        final AtomicBoolean draining = new AtomicBoolean(false);   // 是否有任务在跑
    }

    private final ThreadPoolExecutor pool;
    private final ConcurrentMap<String, DeviceQueue> queues = new ConcurrentHashMap<>();
    private final ConcurrentMap<String, DeviceConnection> conns = new ConcurrentHashMap<>();
    private final AtomicLong rejected = new AtomicLong();

    public AcquisitionScheduler(int parallelism) {
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(parallelism * 64);
        AtomicLong seq = new AtomicLong();
        this.pool = new ThreadPoolExecutor(parallelism, parallelism, 0L, TimeUnit.MILLISECONDS,
                workQueue,
                r -> { Thread t = new Thread(r, "acq-" + seq.incrementAndGet());
                       t.setDaemon(true); return t; },
                new ThreadPoolExecutor.CallerRunsPolicy());   // 池满则由提交方跑 → 反压
    }

    /** 提交一个采集任务；返回的 Future 可按需阻塞或编排。 */
    public <T> CompletableFuture<T> submit(String deviceId, int priority,
                                           long timeoutMs, Command<T> command) {
        Task task = new Task(deviceId, priority, timeoutMs, command);
        DeviceQueue dq = queues.computeIfAbsent(deviceId, k -> new DeviceQueue());
        if (!dq.queue.offer(task)) {                              // 单设备积压上限
            rejected.incrementAndGet();
            task.future.completeExceptionally(
                    new RejectedExecutionException("设备队列已满: " + deviceId));
        } else { pump(dq); }
        @SuppressWarnings("unchecked")
        CompletableFuture<T> f = (CompletableFuture<T>) (CompletableFuture<?>) task.future;
        return f;
    }

    /** CAS 保证同一时刻只有一个线程在消费该设备的队列 → 单设备串行 */
    private void pump(DeviceQueue dq) {
        if (dq.draining.compareAndSet(false, true)) {
            try { pool.execute(() -> drain(dq)); }
            catch (RejectedExecutionException e) {
                dq.draining.set(false);                           // 池满：留给下次提交再泵
                rejected.incrementAndGet();
            }
        }
    }

    private void drain(DeviceQueue dq) {
        try {
            Task t;
            while ((t = dq.queue.poll()) != null) {
                if (System.currentTimeMillis() - t.enqueueTime > t.timeoutMs) {
                    t.future.completeExceptionally(new TimeoutException("排队超时: " + t.deviceId));
                    continue;                                     // 过期任务直接丢弃
                }
                DeviceConnection conn = conns.computeIfAbsent(t.deviceId, DeviceConnection::new);
                try {   // 单次读超时靠 Socket 的 SO_TIMEOUT / Netty ReadTimeoutHandler
                    t.future.complete(t.command.execute(conn));
                } catch (Exception e) {
                    conn.healthy = false;                         // 触发重连
                    t.future.completeExceptionally(e);
                }
            }
        } finally {
            dq.draining.set(false);
            if (!dq.queue.isEmpty()) pump(dq);                    // 期间又来任务 → 再消费一轮
        }
    }

    public int pendingTasks(String deviceId) {
        DeviceQueue dq = queues.get(deviceId);
        return dq == null ? 0 : dq.queue.size();
    }
    public DeviceConnection connection(String deviceId) { return conns.get(deviceId); }
    public long rejectedCount() { return rejected.get(); }

    @Override public void close() { pool.shutdown(); }

    // ---------------- 演示：8 台设备 × 8 个任务 ----------------
    public static void main(String[] args) throws Exception {
        try (AcquisitionScheduler s = new AcquisitionScheduler(8)) {
            java.util.List<CompletableFuture<String>> fs = new java.util.ArrayList<>();
            for (int i = 0; i < 64; i++) {
                String device = "plc-" + (i % 8);
                int idx = i;
                fs.add(s.submit(device, idx % 3, 5_000L,
                        conn -> "device=" + conn.deviceId + " value=" + idx));
            }
            int ok = 0, fail = 0;
            for (CompletableFuture<String> f : fs) {   // 真实场景：编排或异步回调
                try { f.get(10, TimeUnit.SECONDS); ok++; } catch (Exception e) { fail++; }
            }
            System.out.println("成功=" + ok + " 失败=" + fail + " 拒绝=" + s.rejectedCount());
        }
    }
}

代码之外同样重要的配套设计：

1. 定时触发：用 ScheduledExecutorService 按周期 submit 任务，而不是在采集线程里 sleep——周期抖动需要被测量而不是被制造。
2. 批量合并：一个设备一个周期只提交合并后的少量请求（见 10.1）。
3. 优先级：心跳/控制指令 priority=0，周期采集 priority=5，历史补采 priority=9；写操作建议单独队列且不允许批量合并。
4. 三层超时：排队超时（调度器 timeoutMs）、单次读超时（SO_TIMEOUT / ReadTimeoutHandler）、整帧截止时间（见 01 篇 15.1）。
5. 失败处理：连续失败 N 次 → 标记离线 → 关闭连接（不复用）→ 退避重连；重连成功前该设备任务应快速失败而不是一直排队。
6. 指标：每设备的 成功数/失败数/排队长度/最近 RTT/采集周期实际值，这是现场排障的唯一依据。

11. 性能测试方法

  单连接串行吞吐的天花板（先算清，别定不切实际的目标）：
    RTT=2ms → 500 req/s ；RTT=20ms → 50 req/s ；RTT=50ms → 20 req/s
  超天花板只有三条路：批量合并请求 / 多连接（受 PLC 连接上限）/ 流水线（有风险）

# 1) 先用 iperf3 确认网络本身没病
iperf3 -s                                   # 服务端
iperf3 -c 192.168.1.10 -t 60 -i 1           # TCP 带宽（应接近线速）
iperf3 -c 192.168.1.10 -u -b 100M -t 60     # UDP：看 jitter 与丢包率（应接近 0 / <1ms）
# 若 UDP 丢包 > 0 或 jitter 大 → 先修网络/换网线/查交换机端口，别优化代码

# 2) JVM 侧观测
jcmd <pid> VM.native_memory summary         # 直接内存（Netty 堆外）
jstat -gcutil <pid> 1000                    # GC 是否频繁
jstack <pid> | grep -c "acq-"               # 采集线程数

压测的三个注意事项：

1. 闭环 vs 开环："发一个等一个"（闭环）会掩盖问题——请求速率随系统变慢而下降；真实采集是开环（固定周期触发），压测也要按固定速率发，否则低估积压与长尾（coordinated omission）。
2. 先用模拟器排除设备瓶颈：用 Modbus/S7 从站仿真工具测采集器自身极限，再接真实 PLC，否则分不清瓶颈在采集器还是 PLC 的通信处理器。
3. 别在采集器本机压测（压测客户端会抢 CPU 与网络）；延迟用 HdrHistogram / Prometheus histogram 记录，不要用平均值——平均值会被大量快请求掩盖长尾，而长尾正是采集周期抖动的来源。

小结

• 五种 IO 模型的关键区分是"等待数据"与"拷贝数据"两阶段是否阻塞；只有 AIO 两阶段都不阻塞，其余（含 IO 多路复用）都是同步 IO。
• select 有 1024 限制且 O(n) 轮询；poll 去掉数量限制但仍 O(n)；epoll 用红黑树 + 就绪链表做到 O(1)，支持 LT/ET 与 EPOLLONESHOT。
• ET 必须配非阻塞 fd 且一次读到 EAGAIN，漏读后再也不通知。
• BIO 的"每设备一线程"在工业场景合理：设备数百级、连接长期持有、单连接利用率不到 1%，200 线程的代价远小于手写 NIO 的 bug 成本。
• 手写 NIO 的痛点：拆包、写半包、Buffer 状态、Selector 空轮询、连接管理 → 生产用 Netty。
• Linux 上 AIO 由线程池 + epoll 模拟，不如 NIO；Windows 的 IOCP 才是真异步。
• Reactor 三种形态：单/单、单/多、主从多（Netty 采用）。一个连接的所有事件固定在同一线程内串行执行，是无锁设计的基础。
• Netty：boss 负责 accept、worker 负责 IO；Channel 终身绑定一个 EventLoop → 串行无锁；Pipeline 双向链表，inbound 从头到尾、outbound 从尾到头；ByteBuf 的引用计数必须配对 release()。
• 零拷贝：传统 read+write 是 4 次拷贝 4 次切换；sendfile（SG-DMA）降到 2 次纯 DMA 拷贝；Java 侧是 FileChannel.transferTo 与 MappedByteBuffer。
• 线程池：无界队列会在批量掉线时 OOM 并在恢复瞬间二次冲击 PLC；用有界队列 + CallerRunsPolicy；线程数由"并行设备数"驱动；虚拟线程让同步写法几乎免费，但要避开 synchronized pinning。
• 工业并发核心约束：PLC 连接数是硬上限（S7-1200 只有 8 个），连接长期持有而非池化；一条连接必须串行（否则字节交错 + 响应错配，后者是"数据看着正常但值是错的"这类最危险的 bug）；事务 ID 流水线风险大，默认串行、写操作绝不流水。
• 优化顺序：先批量合并请求（数量级收益），再谈 IO 模型（百分比收益）。
• 压测：单连接 TPS 上限 = 1000/RTT(ms)；iperf3 先排除网络问题；用 P95/P99 与直方图而非平均值；按开环固定速率压测。

常见误区

本文是《工业通信与数据采集》栏目第 6/22 篇 · 上一篇：粘包与拆包 ｜ 栏目总览 ｜ 下一篇：Modbus 协议详解