IO 模型：BIO、NIO 与 Reactor 模式

public class BioServer {
    public static void main(String[] args) throws Exception {
        ServerSocket server = new ServerSocket(8080);
        System.out.println("BIO Server 启动，端口 8080");

        while (true) {
            // ★ accept() 会阻塞，直到有客户端连接
            Socket socket = server.accept();

            // ★ 每个连接必须启动一个新线程处理
            new Thread(() -> {
                try (InputStream in = socket.getInputStream();
                     OutputStream out = socket.getOutputStream()) {

                    byte[] buf = new byte[1024];
                    int len;
                    // ★ read() 会阻塞，直到有数据可读
                    while ((len = in.read(buf)) != -1) {
                        out.write(buf, 0, len);  // echo 回写
                        out.flush();
                    }
                } catch (Exception e) { e.printStackTrace(); }
            }).start();
        }
    }
}

public class NioServer {
    public static void main(String[] args) throws Exception {
        Selector selector = Selector.open();

        ServerSocketChannel serverChannel = ServerSocketChannel.open();
        serverChannel.bind(new InetSocketAddress(8080));
        serverChannel.configureBlocking(false);  // ★ 非阻塞模式

        // 将 ServerChannel 注册到 Selector，监听 ACCEPT 事件
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);

        ByteBuffer buffer = ByteBuffer.allocate(1024);

        while (true) {
            // ★ select() 阻塞直到有事件就绪
            selector.select();

            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> it = keys.iterator();

            while (it.hasNext()) {
                SelectionKey key = it.next();
                it.remove();  // ★ 必须移除，否则下次会重复处理

                if (key.isAcceptable()) {
                    SocketChannel client = serverChannel.accept();
                    client.configureBlocking(false);
                    client.register(selector, SelectionKey.OP_READ);
                } else if (key.isReadable()) {
                    SocketChannel client = (SocketChannel) key.channel();
                    buffer.clear();
                    int len = client.read(buffer);  // ★ 非阻塞，可能返回 0
                    if (len > 0) {
                        buffer.flip();  // ★ 切换为读模式
                        client.write(buffer);
                    }
                }
            }
        }
    }
}

ByteBuffer buf = ByteBuffer.allocate(10);
// capacity=10, position=0, limit=10

// 写入 "Hello"（5 个字节）
buf.put("Hello".getBytes());
// capacity=10, position=5, limit=10

// flip()：准备读取——limit 设为当前 position，position 归零
buf.flip();
// capacity=10, position=0, limit=5

// 读取 5 字节
byte[] data = new byte[buf.remaining()];  // remaining = limit - position = 5
buf.get(data);
// capacity=10, position=5, limit=5

// clear()：重置为写模式
buf.clear();
// capacity=10, position=0, limit=10（数据还在，但 position 重置了）

// compact()：丢弃已读数据，保留未读数据
buf.compact();
// 将未读数据移到缓冲区开头，position 紧跟其后

/* ═══════════ 模型一：单线程 Reactor ═══════════
 *
 *   Acceptor/Reader/Writer 全部在一个线程中
 *
 *   [Reactor Thread]
 *     ├── select() 监听事件
 *     ├── accept() 处理连接
 *     ├── read()   读取数据
 *     ├── process() 业务处理 ← 阻塞在这里会影响其他事件！
 *     └── write()  写回响应
 *
 *   优点：简单，无线程切换开销
 *   缺点：业务处理耗时会阻塞整个 Reactor
 *   适用：业务逻辑简单、处理速度快的场景
 */

/* ═══════════ 模型二：多线程 Reactor（一主多从）═══════════
 *
 *   Reactor 线程只负责 IO，业务处理交给 Worker 线程池
 *
 *   [Reactor Thread]            [Worker Pool]
 *     ├── select()                ├── Thread-1: process()
 *     ├── accept()                ├── Thread-2: process()
 *     ├── read()                  └── Thread-3: process()
 *     └── write() ←── 回调
 *
 *   优点：IO 和业务分离，互不阻塞
 *   缺点：单 Reactor 仍是连接数瓶颈（百万连接时 select 压力大）
 *   适用：中等并发，业务处理较重的场景
 */

/* ═══════════ 模型三：主从 Reactor（多线程 + 多 Reactor）═══════════
 *
 *   MainReactor 负责 accept，SubReactor 负责读写
 *
 *   [MainReactor]
 *     └── accept() → 分配给 SubReactor
 *
 *   [SubReactor-1]     [SubReactor-2]     [SubReactor-N]
 *     ├── select()       ├── select()       ├── select()
 *     ├── read()         ├── read()         ├── read()
 *     └── write()        └── write()        └── write()
 *          ↓                  ↓                  ↓
 *   [Worker Pool]（如果业务处理也很重）
 *
 *   优点：连接建立和 IO 读写分离，支持百万连接
 *   缺点：实现复杂
 *   适用：高并发、高性能场景（Netty 采用此模型）
 */

public class NettyServer {
    public void start() throws Exception {
        // BossGroup：MainReactor，负责 accept 新连接（通常 1 个线程）
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);

        // WorkerGroup：SubReactor，负责已连接 Channel 的 IO 读写（默认 CPU*2 线程）
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    ChannelPipeline p = ch.pipeline();
                    p.addLast(new Decoder());       // 解码器
                    p.addLast(new Encoder());       // 编码器
                    p.addLast(new BusinessHandler()); // 业务逻辑
                }
            })
            .option(ChannelOption.SO_BACKLOG, 1024)   // 连接队列大小
            .childOption(ChannelOption.SO_KEEPALIVE, true);

        // 绑定端口，异步启动
        ChannelFuture future = bootstrap.bind(8080).sync();
        future.channel().closeFuture().sync();

        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}

/*
 * Netty 的 ChannelPipeline 是一条 Handler 链（责任链模式）：
 *
 *  Inbound:  网络数据 → Decoder → Handler1 → Handler2 → ...
 *  Outbound: ... ← Handler2 ← Handler1 ← Encoder ← 业务写入
 *
 * 关键设计：
 * ① 每个 Channel 有独立的 Pipeline，线程安全
 * ② 同一个 Channel 的所有操作都在同一个 EventLoop 线程执行，无需加锁
 * ③ 如果 Handler 中有耗时操作，必须提交到独立的业务线程池
 */

public class BusinessHandler extends SimpleChannelInboundHandler<String> {
    // ★ 业务线程池——耗时操作不在 EventLoop 中执行
    private static final EventExecutorGroup bizGroup =
        new DefaultEventExecutorGroup(16);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        // 如果业务处理很快，直接在 EventLoop 线程执行即可
        // 如果涉及 DB 查询、RPC 调用等耗时操作，必须异步
        ctx.writeAndFlush("Echo: " + msg);
    }
}

// 或者在 pipeline 中指定业务线程池：
pipeline.addLast(bizGroup, "biz", new BusinessHandler());

public class AioServer {
    public static void main(String[] args) throws Exception {
        AsynchronousServerSocketChannel server =
            AsynchronousServerSocketChannel.open().bind(new InetSocketAddress(8080));

        System.out.println("AIO Server 启动");

        // ★ 异步 accept：发起后立即返回，连接到达时回调 completed()
        server.accept(null, new CompletionHandler<AsynchronousSocketChannel, Void>() {
            @Override
            public void completed(AsynchronousSocketChannel client, Void att) {
                // 继续接受下一个连接
                server.accept(null, this);

                ByteBuffer buf = ByteBuffer.allocate(1024);
                // ★ 异步 read：发起后立即返回，数据就绪时回调
                client.read(buf, buf, new CompletionHandler<Integer, ByteBuffer>() {
                    @Override
                    public void completed(Integer len, ByteBuffer buffer) {
                        buffer.flip();
                        client.write(buffer);
                    }
                    @Override
                    public void failed(Throwable exc, ByteBuffer buffer) { }
                });
            }
            @Override
            public void failed(Throwable exc, Void att) { }
        });

        System.in.read(); // 主线程阻塞等待
    }
}