// 原生 NIO EchoServer 的核心循环（看似正确，实则隐患重重）
while (true) {
    selector.select();
    for (SelectionKey key : selector.selectedKeys()) {
        if (key.isAcceptable()) {
            SocketChannel client = ssc.accept();
            client.configureBlocking(false);
            client.register(selector, SelectionKey.OP_READ);
        } else if (key.isReadable()) {
            SocketChannel client = (SocketChannel) key.channel();
            ByteBuffer buf = ByteBuffer.allocate(1024);
            int len = client.read(buf);   // 问题1: 读到半包怎么办？
            buf.flip();                    // 问题2: 忘了 flip 直接炸
            client.write(buf);             // 问题3: 写不出去谁管？
            buf.clear();                   // 问题4: clear 还是 compact？
        }
        keyIterator.remove();
    }
}

// Netty 源码简化示意: io.netty.channel.nio.NioEventLoop
public final class NioEventLoop extends SingleThreadEventLoop {
    // 每个 EventLoop 持有独立的 Selector (内部是 epoll 实例)
    private Selector selector;
    // 关联的线程
    private volatile Thread thread;
    // 任务队列：外部线程提交的任务存在这里
    private final Queue<Runnable> taskQueue;

    @Override
    protected void run() {
        for (;;) {
            // 1. select(): 检查就绪 Channel (底层 epoll_wait)
            selector.select(timeoutMillis);
            // 2. processSelectedKeys(): 处理 I/O 事件
            processSelectedKeys();
            // 3. runAllTasks(): 执行任务队列中的 Runnable
            runAllTasks();
        }
    }
}

// 创建两个 EventLoopGroup
EventLoopGroup bossGroup   = new NioEventLoopGroup(1);     // boss 组，1 个线程即可
EventLoopGroup workerGroup = new NioEventLoopGroup();      // worker 组，默认 CPU核数*2

ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)                          // ① 绑定两个线程组
         .channel(NioServerSocketChannel.class)                   // ② 指定 Channel 类型
         .option(ChannelOption.SO_BACKLOG, 128)                   // ③ TCP 参数
         .childOption(ChannelOption.SO_KEEPALIVE, true)           // ④ 客户端 Channel 参数
         .childHandler(new ChannelInitializer<SocketChannel>() {  // ⑤ 客户端 Channel 处理器
             @Override
             protected void initChannel(SocketChannel ch) {
                 // 添加 Handler 到 Pipeline
                 ch.pipeline().addLast(new MyHandler());
             }
         });

ChannelFuture future = bootstrap.bind(8080).sync();  // 绑定端口，启动服务

.childHandler(new ChannelInitializer<SocketChannel>() {
    @Override
    protected void initChannel(SocketChannel ch) {
        ChannelPipeline p = ch.pipeline();
        p.addLast(new StringDecoder());       // 解码: ByteBuf → String
        p.addLast(new StringEncoder());       // 编码: String → ByteBuf
        p.addLast(new MyBusinessHandler());   // 业务逻辑
    }
});

// 使用 SimpleChannelInboundHandler —— 最常用的业务 Handler 写法
public class EchoHandler extends SimpleChannelInboundHandler<ByteBuf> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
        // msg 是解码后的消息，方法结束后自动释放
        ctx.writeAndFlush(msg);  // 回显
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println("Client connected: " + ctx.channel().remoteAddress());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        System.out.println("Client disconnected: " + ctx.channel().remoteAddress());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();  // 发生异常时关闭连接
    }
}

// 创建 ByteBuf
ByteBuf buf = Unpooled.buffer(256);         // 非池化，初始容量 256

// 写入（writerIndex 自动前进）
buf.writeInt(42);                           // 写入 int（4 字节）
buf.writeBytes("Hello".getBytes());         // 写入字节数组

// 读取（readerIndex 自动前进）
int value = buf.readInt();                  // 读取 int（如果可读字节不够抛异常）
byte[] bytes = new byte[buf.readableBytes()];
buf.readBytes(bytes);                       // 读取所有可读字节

// 查询（不移动指针）
buf.getByte(0);                             // 按绝对位置读取，不影响 readerIndex

// 标记与回退
buf.markReaderIndex();                      // 标记当前读位置
buf.resetReaderIndex();                     // 回退到标记的读位置

// 丢弃已读数据（compact）
buf.discardReadBytes();                     // 将可读区域移到开头，释放已读空间

// 引用计数
buf.retain();                               // 引用计数 +1
buf.release();                              // 引用计数 -1，归零后释放内存

// LengthFieldBasedFrameDecoder 参数详解
new LengthFieldBasedFrameDecoder(
    1024 * 1024,   // maxFrameLength: 最大帧长度，超过则抛异常
    0,             // lengthFieldOffset: 长度字段偏移量（从第几个字节开始）
    4,             // lengthFieldLength: 长度字段占几个字节
    0,             // lengthAdjustment: 长度调整值
    4              // initialBytesToStrip: 解码后剥离前几个字节
);

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

/**
 * Netty Echo Server —— 最简示例。
 * 启动后监听 8080 端口，收到任何数据原样返回。
 * 测试: telnet localhost 8080
 */
public class NettyEchoServer {
    private final int port;

    public NettyEchoServer(int port) {
        this.port = port;
    }

    public void start() throws InterruptedException {
        EventLoopGroup bossGroup   = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) {
                     ch.pipeline().addLast(new EchoHandler());
                 }
             });

            ChannelFuture f = b.bind(port).sync();
            System.out.println("Echo Server started on port " + port);
            f.channel().closeFuture().sync(); // 阻塞直到服务端关闭
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    /** 业务 Handler：收到 ByteBuf 直接回写 */
    private static class EchoHandler extends SimpleChannelInboundHandler<ByteBuf> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
            // retain() 增加引用计数，确保 msg 在 writeAndFlush 完成前不被释放
            ctx.writeAndFlush(msg.retain());
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace();
            ctx.close();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new NettyEchoServer(8080).start();
    }
}

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * Netty Time Server —— 连接后返回当前时间字符串。
 * 测试: telnet localhost 8080
 */
public class NettyTimeServer {
    private final int port;

    public NettyTimeServer(int port) {
        this.port = port;
    }

    public void start() throws InterruptedException {
        EventLoopGroup bossGroup   = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) {
                     ChannelPipeline p = ch.pipeline();
                     // 解码器：ByteBuf → String（入站）
                     p.addLast(new StringDecoder());
                     // 编码器：String → ByteBuf（出站）
                     p.addLast(new StringEncoder());
                     // 业务 Handler（入站）：接收 String 类型消息
                     p.addLast(new TimeServerHandler());
                 }
             });

            ChannelFuture f = b.bind(port).sync();
            System.out.println("Time Server started on port " + port);
            f.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    /**
     * Handler 处理 String 消息（已由 StringDecoder 解码）。
     * 收到任意消息 → 返回当前时间字符串 → 关闭连接。
     */
    private static class TimeServerHandler extends SimpleChannelInboundHandler<String> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) {
            String now = LocalDateTime.now()
                .format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            ctx.writeAndFlush("Server Time: " + now + "\r\n");
            ctx.close(); // 发送时间后关闭连接
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace();
            ctx.close();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new NettyTimeServer(8080).start();
    }
}

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

/**
 * Netty Time Client —— 连接服务器，接收时间字符串后打印。
 */
public class NettyTimeClient {
    private final String host;
    private final int port;

    public NettyTimeClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void start() throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
             .channel(NioSocketChannel.class)
             .handler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) {
                     ch.pipeline().addLast(new StringDecoder());
                     ch.pipeline().addLast(new StringEncoder());
                     ch.pipeline().addLast(new TimeClientHandler());
                 }
             });

            ChannelFuture f = b.connect(host, port).sync();
            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }

    /** 客户端 Handler：连接建立后不做操作，等待服务器发送时间 */
    private static class TimeClientHandler extends SimpleChannelInboundHandler<String> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) {
            System.out.println("Received: " + msg);  // 打印服务器返回的时间
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace();
            ctx.close();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new NettyTimeClient("localhost", 8080).start();
    }
}

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.concurrent.GlobalEventExecutor;

/**
 * Netty 群聊服务器 —— 任一客户端发送消息，广播给所有在线客户端。
 * 测试: 用多个 telnet 连接 8080，在其中输入文字，其他窗口都会收到。
 */
public class NettyChatServer {
    private final int port;

    public NettyChatServer(int port) {
        this.port = port;
    }

    public void start() throws InterruptedException {
        EventLoopGroup bossGroup   = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) {
                     ch.pipeline().addLast(new StringDecoder());
                     ch.pipeline().addLast(new StringEncoder());
                     ch.pipeline().addLast(new ChatHandler());
                 }
             });

            ChannelFuture f = b.bind(port).sync();
            System.out.println("Chat Server started on port " + port);
            f.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    /** 群聊 Handler：维护 ChannelGroup，实现消息广播 */
    private static class ChatHandler extends SimpleChannelInboundHandler<String> {
        // ChannelGroup 是线程安全的 Channel 集合
        private static final ChannelGroup channels =
            new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            Channel incoming = ctx.channel();
            // 广播上线消息给所有人（包括自己）
            channels.writeAndFlush("[SYSTEM] " + incoming.remoteAddress() + " joined.\r\n");
            channels.add(incoming); // 加入群组
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) {
            Channel sender = ctx.channel();
            // 广播消息给所有人（除发送者外）
            for (Channel ch : channels) {
                if (ch != sender) {
                    ch.writeAndFlush("[" + sender.remoteAddress() + "] " + msg + "\r\n");
                } else {
                    ch.writeAndFlush("[You] " + msg + "\r\n");
                }
            }
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) {
            Channel leaving = ctx.channel();
            channels.remove(leaving); // 从群组移除
            channels.writeAndFlush("[SYSTEM] " + leaving.remoteAddress() + " left.\r\n");
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace();
            ctx.close();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new NettyChatServer(8080).start();
    }
}

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.util.CharsetUtil;

import static io.netty.handler.codec.http.HttpHeaderNames.*;

/**
 * Netty HTTP Server —— 处理 GET 请求，返回 JSON 格式的 Hello World。
 * 浏览器访问: http://localhost:8080
 */
public class NettyHttpServer {
    private final int port;

    public NettyHttpServer(int port) {
        this.port = port;
    }

    public void start() throws InterruptedException {
        EventLoopGroup bossGroup   = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) {
                     ch.pipeline()
                       // HttpRequestDecoder: 将 HTTP 请求字节流解码为 HttpRequest 对象
                       .addLast(new HttpServerCodec())
                       // HttpObjectAggregator: 将 HTTP 消息的多个部分聚合成完整的 FullHttpRequest
                       .addLast(new HttpObjectAggregator(65536))
                       // 业务 Handler
                       .addLast(new HttpServerHandler());
                 }
             });

            ChannelFuture f = b.bind(port).sync();
            System.out.println("HTTP Server started on http://localhost:" + port);
            f.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    /** HTTP 请求处理器 */
    private static class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) {
            String uri = request.uri();

            // 构建响应内容
            String responseBody = "{\"message\": \"Hello from Netty\", \"uri\": \"" + uri + "\"}";

            // 构建 HTTP 响应
            FullHttpResponse response = new DefaultFullHttpResponse(
                HttpVersion.HTTP_1_1,
                HttpResponseStatus.OK,
                Unpooled.copiedBuffer(responseBody, CharsetUtil.UTF_8)
            );

            response.headers()
                .set(CONTENT_TYPE, "application/json; charset=UTF-8")
                .set(CONTENT_LENGTH, response.content().readableBytes());

            // 发送响应
            ctx.writeAndFlush(response)
               .addListener(ChannelFutureListener.CLOSE); // 发送完成后关闭连接
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace();
            ctx.close();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new NettyHttpServer(8080).start();
    }
}

// Netty 源码简化示意: io.netty.channel.nio.NioEventLoop
// select() 的超时判断逻辑

long currentTimeNanos = System.nanoTime();
for (;;) {
    long timeoutMillis = delayNanos / 1000000L;
    if (timeoutMillis <= 0) {
        selector.selectNow();  // 无超时，非阻塞
        break;
    }
    // select(timeout) 返回了就绪的 Channel 数量
    int selectedKeys = selector.select(timeoutMillis);
    selectCnt++;  // select 操作计数

    if (selectedKeys != 0) {
        break;    // 正常：有 Channel 就绪
    }

    // 下面是 Netty 的"空轮询检测"逻辑
    long time = System.nanoTime();
    if (time - currentTimeNanos >= timeoutMillis) {
        selectCnt = 1;  // 正常：超时时间内确实没有事件，重置计数
    } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
               selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
        // 检测到空轮询！select 返回了 0 个 Channel，但实际时间远小于超时时间
        logger.warn("Selector.select() returned prematurely {} times in a row; "
                    + "rebuilding Selector.", selectCnt);
        rebuildSelector();  // 重新创建 Selector
        selectCnt = 1;
        break;
    }
}