Netty 与高性能网络模型

EventLoop (接口)
    └─ OrderedEventExecutor (接口,保证任务顺序执行)
        └─ EventExecutor (接口,事件执行器)
            └─ AbstractEventExecutor (抽象类)
                └─ AbstractScheduledEventExecutor (支持定时任务)
                    └─ SingleThreadEventExecutor (单线程执行器)
                        └─ SingleThreadEventLoop (单线程事件循环)
                            └─ NioEventLoop (NIO 实现)
                            └─ EpollEventLoop (Linux epoll 实现)

// NioEventLoop 的核心执行逻辑
@Override
protected void run() {
    int selectCnt = 0;
    for (;;) {
        try {
            int strategy;
            try {
                // 1. 计算选择策略(是否有任务需要执行)
                strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());
                switch (strategy) {
                    case SelectStrategy.CONTINUE:
                        continue;
                    case SelectStrategy.BUSY_WAIT:
                        // fall through to SELECT since busy wait is not supported
                    case SelectStrategy.SELECT:
                        // 2. 阻塞等待 IO 事件
                        long curDeadlineNanos = nextScheduledTaskDeadlineNanos();
                        if (curDeadlineNanos == -1L) {
                            curDeadlineNanos = NONE; // nothing on the calendar
                        }
                        nextWakeupNanos.set(curDeadlineNanos);
                        try {
                            if (!hasTasks()) {
                                strategy = select(curDeadlineNanos);
                            }
                        } finally {
                            // This timeout is applied to correct the timeout of select,
                            // so this needs to be reset every time.
                            nextWakeupNanos.set(AWAKE);
                        }
                        // fall through
                    default:
                }
            } catch (IOException e) {
                // 处理异常
                rebuildSelector0();
                selectCnt = 0;
                handleLoopException(e);
                continue;
            }

            // 3. 处理 IO 事件
            selectCnt++;
            cancelledKeys = 0;
            needsToSelectAgain = false;
            final int ioRatio = this.ioRatio;
            boolean ranTasks;
            if (ioRatio == 100) {
                try {
                    if (strategy > 0) {
                        processSelectedKeys();
                    }
                } finally {
                    // 4. 执行所有任务
                    ranTasks = runAllTasks();
                }
            } else {
                final long ioStartTime = System.nanoTime();
                try {
                    if (strategy > 0) {
                        processSelectedKeys();
                    }
                } finally {
                    final long ioTime = System.nanoTime() - ioStartTime;
                    // 5. 根据 ioRatio 控制任务执行时间
                    ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                }
            }

            if (ranTasks || strategy > 0) {
                if (selectCnt >= MIN_PREMATURE_SELECTOR_RETURNS && logger.isDebugEnabled()) {
                    logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.",
                            selectCnt - 1, selector);
                }
                selectCnt = 0;
            } else if (unexpectedSelectorWakeup(selectCnt)) { // Unexpected wakeup
                selectCnt = 0;
            }
        } catch (CancelledKeyException e) {
            // 处理异常
        } catch (Error e) {
            throw (Error) e;
        } catch (Throwable t) {
            handleLoopException(t);
        } finally {
            // Always handle shutdown even if the loop processing threw an exception.
            try {
                if (isShuttingDown()) {
                    closeAll();
                    if (confirmShutdown()) {
                        return;
                    }
                }
            } catch (Error e) {
                throw (Error) e;
            } catch (Throwable t) {
                handleLoopException(t);
            }
        }
    }
}

// 推荐配置
EventLoopGroup bossGroup = new NioEventLoopGroup(1); // boss 只需 1 个线程
EventLoopGroup workerGroup = new NioEventLoopGroup(); // 默认 CPU 核心数 * 2

// 高性能配置 (Linux 环境)
EventLoopGroup bossGroup = new EpollEventLoopGroup(1);
EventLoopGroup workerGroup = new EpollEventLoopGroup();

// 线程数计算公式
int threadCount = Math.max(1, SystemPropertyUtil.getInt(
    "io.netty.eventLoopThreads", 
    NettyRuntime.availableProcessors() * 2
));

Channel (接口)
    └─ AttributeMap (接口,属性映射)
        └─ AbstractChannel (抽象类)
            ├─ AbstractServerChannel (服务端通道基类)
            │   ├─ LocalServerChannel (本地通信)
            │   ├─ NioServerSocketChannel (NIO 服务端)
            │   └─ EpollServerSocketChannel (Linux epoll 服务端)
            └─ AbstractClientChannel (客户端通道基类)
                ├─ LocalChannel (本地通信)
                ├─ NioSocketChannel (NIO 客户端)
                └─ EpollSocketChannel (Linux epoll 客户端)

public interface Channel extends AttributeMap, Comparable<Channel> {
    // 获取 EventLoop
    EventLoop eventLoop();
    
    // 获取父 Channel (服务端 Channel 的子 Channel)
    Channel parent();
    
    // 获取 Channel 配置
    ChannelConfig config();
    
    // 判断是否打开
    boolean isOpen();
    
    // 判断是否已注册
    boolean isRegistered();
    
    // 判断是否活跃 (已连接)
    boolean isActive();
    
    // 获取本地地址
    SocketAddress localAddress();
    
    // 获取远程地址
    SocketAddress remoteAddress();
    
    // 获取 Pipeline
    ChannelPipeline pipeline();
    
    // 获取 Unsafe (内部使用)
    Unsafe unsafe();
    
    // 异步绑定地址
    ChannelFuture bind(SocketAddress localAddress);
    
    // 异步连接远程地址
    ChannelFuture connect(SocketAddress remoteAddress);
    
    // 异步断开连接
    ChannelFuture disconnect();
    
    // 异步关闭
    ChannelFuture close();
    
    // 异步注销
    ChannelFuture deregister();
    
    // 同步写入消息
    ChannelFuture writeAndFlush(Object msg);
}

// Channel 状态流转图
@Sharable
public class ChannelStateHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) {
        // 状态 1: Channel 已注册到 EventLoop
        System.out.println("Channel 已注册: " + ctx.channel());
        ctx.fireChannelRegistered();
    }
    
    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) {
        // 状态 2: Channel 从 EventLoop 注销
        System.out.println("Channel 已注销: " + ctx.channel());
        ctx.fireChannelUnregistered();
    }
    
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        // 状态 3: Channel 已激活 (可以读写)
        System.out.println("Channel 已激活: " + ctx.channel().remoteAddress());
        ctx.fireChannelActive();
    }
    
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        // 状态 4: Channel 已失活 (连接断开)
        System.out.println("Channel 已失活: " + ctx.channel().remoteAddress());
        ctx.fireChannelInactive();
    }
}

// 状态转换图
/*
    [未注册] --register()--> [已注册] --channelActive()--> [活跃]
       ↑                        ↓                              ↓
       |                    channelInactive()              disconnect()
       |                        ↓                              ↓
    [已注销] <--unregister()-- [已注册] <--channelInactive()-- [失活]
*/

    I/O Request
       │
       ▼
┌──────────────────────────────────────────────────┐
│              ChannelPipeline                      │
│                                                   │
│  ┌─────────┬─────────┬─────────┬─────────┐      │
│  │ Handler │ Handler │ Handler │ Handler │      │
│  │   (1)   │   (2)   │   (3)   │   (4)   │      │
│  │ Inbound │ Inbound │Outbound │Outbound │      │
│  └────┬────┴────┬────┴────┬────┴────┬────┘      │
│       │         │         │         │            │
│       ▼         ▼         ▼         ▼            │
│  Inbound ←─── Inbound ←──│──→ Outbound → Outbound│
│  Head ───────────────────┼─────────────────── Tail│
└──────────────────────────────────────────────────┘
       │         ▲                   │
       ▼         │                   ▼
   SocketRead    │              SocketWrite
                 │
            TailContext
         (自动释放资源)

public class DefaultChannelPipeline implements ChannelPipeline {
    // 头节点和尾节点
    final AbstractChannelHandlerContext head;
    final AbstractChannelHandlerContext tail;
    
    // 添加 Handler
    @Override
    public final ChannelPipeline addLast(String name, ChannelHandler handler) {
        return addLast(null, name, handler);
    }
    
    @Override
    public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
        final AbstractChannelHandlerContext newCtx;
        synchronized (this) {
            // 1. 检查是否重复添加非 @Sharable 的 Handler
            checkMultiplicity(handler);
            
            // 2. 创建 Context 节点
            newCtx = newContext(group, filterName(name, handler), handler);
            
            // 3. 添加到链表尾部
            addLast0(newCtx);
            
            // 4. 如果 Channel 未注册,设置为待添加状态
            if (!registered) {
                newCtx.setAddPending();
                callHandlerCallbackLater(newCtx, true);
                return this;
            }
            
            // 5. 回调 handlerAdded
            callHandlerAdded0(newCtx);
        }
        return this;
    }
    
    private void addLast0(AbstractChannelHandlerContext newCtx) {
        AbstractChannelHandlerContext prev = tail.prev;
        newCtx.prev = prev;
        newCtx.next = tail;
        prev.next = newCtx;
        tail.prev = newCtx;
    }
}

// 服务端 Pipeline 典型配置
channel.pipeline()
    // ===== 入站处理器 (从前到后) =====
    .addLast("frameDecoder", new LengthFieldBasedFrameDecoder(1024, 0, 4, 0, 4))
    .addLast("stringDecoder", new StringDecoder(StandardCharsets.UTF_8))
    .addLast("idleStateHandler", new IdleStateHandler(60, 0, 0))
    .addLast("heartbeatHandler", new HeartbeatHandler())
    .addLast("businessHandler", new BusinessHandler())
    
    // ===== 出站处理器 (从后到前) =====
    .addLast("stringEncoder", new StringEncoder(StandardCharsets.UTF_8))
    .addLast("frameEncoder", new LengthFieldPrepender(4))
    .addLast("exceptionHandler", new ExceptionHandler());

ChannelHandler (接口)
    ├─ ChannelInboundHandler (入站接口)
    │   └─ ChannelInboundHandlerAdapter (入站适配器)
    │       ├─ SimpleChannelInboundHandler<I> (自动释放消息)
    │       ├─ ByteToMessageDecoder (字节解码器)
    │       │   ├─ LineBasedFrameDecoder (换行符拆包)
    │       │   ├─ DelimiterBasedFrameDecoder (分隔符拆包)
    │       │   ├─ LengthFieldBasedFrameDecoder (长度字段拆包)
    │       │   └─ ReplayingDecoder<S> (可重放解码器)
    │       └─ MessageToMessageDecoder<I> (消息到消息解码器)
    │
    └─ ChannelOutboundHandler (出站接口)
        └─ ChannelOutboundHandlerAdapter (出站适配器)
            ├─ MessageToByteEncoder<I> (消息编码器)
            │   └─ LengthFieldPrepender (长度字段编码)
            └─ MessageToMessageEncoder<I> (消息到消息编码器)

public interface ChannelHandler {
    // Handler 被添加到 Pipeline 时调用
    void handlerAdded(ChannelHandlerContext ctx) throws Exception;
    
    // Handler 从 Pipeline 移除时调用
    void handlerRemoved(ChannelHandlerContext ctx) throws Exception;
    
    // 标记该 Handler 是否可被多个 Pipeline 共享
    @Deprecated
    void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception;
    
    // Sharable 注解
    @Inherited
    @Documented
    @Target(ElementType.TYPE)
    @Retention(RetentionPolicy.RUNTIME)
    @interface Sharable {
        // 标记 Handler 可以被多个 Pipeline 共享
    }
}

// 入站 Handler 生命周期
public interface ChannelInboundHandler extends ChannelHandler {
    void channelRegistered(ChannelHandlerContext ctx) throws Exception;
    void channelUnregistered(ChannelHandlerContext ctx) throws Exception;
    void channelActive(ChannelHandlerContext ctx) throws Exception;
    void channelInactive(ChannelHandlerContext ctx) throws Exception;
    void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception;
    void channelReadComplete(ChannelHandlerContext ctx) throws Exception;
    void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception;
    void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception;
    void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception;
}

// Pipeline 配置
pipeline.addLast("A", new InboundHandlerA());  // 入站
pipeline.addLast("B", new InboundHandlerB());  // 入站
pipeline.addLast("C", new OutboundHandlerC()); // 出站
pipeline.addLast("D", new OutboundHandlerD()); // 出站

// 入站事件传播顺序: A -> B -> C -> D (C 和 D 不处理入站事件)
// 出站事件传播顺序: D -> C -> B -> A (B 和 A 不处理出站事件)

// 典型示例
public class InboundHandlerA extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        System.out.println("InboundHandler A: " + msg);
        ctx.fireChannelRead(msg); // 传播给下一个 InboundHandler
    }
}

public class OutboundHandlerC extends ChannelOutboundHandlerAdapter {
    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
        System.out.println("OutboundHandler C: " + msg);
        ctx.write(msg, promise); // 传播给下一个 OutboundHandler
    }
}

// × 错误示例:编码器放在业务处理器之前
pipeline.addLast("encoder", new StringEncoder());      // 出站
pipeline.addLast("decoder", new StringDecoder());      // 入站
pipeline.addLast("business", new BusinessHandler());   // 入站

// 问题: 写入 String 时,编码器在业务处理器之前,无法拦截处理
// 出站顺序: business(不处理出站) -> decoder(不处理出站) -> encoder

// √ 正确示例:编码器放在业务处理器之后
pipeline.addLast("decoder", new StringDecoder());      // 入站
pipeline.addLast("business", new BusinessHandler());   // 入站
pipeline.addLast("encoder", new StringEncoder());      // 出站

// 出站顺序: encoder -> business(不处理) -> decoder(不处理) -> network

// 在某个 Handler 中不调用 fireXxx 方法,则事件传播终止
public class AuthHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (!isAuthenticated(ctx)) {
            // 认证失败,不传播事件,直接关闭连接
            ctx.writeAndFlush("Authentication failed");
            ctx.close();
            return; // 不调用 fireChannelRead,事件传播终止
        }
        // 认证成功,继续传播
        ctx.fireChannelRead(msg);
    }
}

      +-------------------+------------------+------------------+
      | discardable bytes |  readable bytes  |  writable bytes  |
      |                   |     (content)    |                  |
      +-------------------+------------------+------------------+
      |                   |                  |                  |
      0      <=      readerIndex   <=   writerIndex    <=    capacity

// ByteBuf 核心属性
public abstract class AbstractByteBuf extends ByteBuf {
    int readerIndex;  // 读索引
    int writerIndex;  // 写索引
    int markedReaderIndex;  // 标记的读索引
    int markedWriterIndex;  // 标记的写索引
    int maxCapacity;  // 最大容量
}

public abstract class ByteBuf implements ReferenceCounted, Comparable<ByteBuf> {
    // ===== 读写操作 =====
    
    // 读取数据
    public abstract byte readByte();
    public abstract ByteBuf readBytes(byte[] dst);
    public abstract ByteBuf readBytes(ByteBuf dst);
    public abstract int readInt();
    public abstract long readLong();
    
    // 写入数据
    public abstract ByteBuf writeByte(int value);
    public abstract ByteBuf writeBytes(byte[] src);
    public abstract ByteBuf writeBytes(ByteBuf src);
    public abstract ByteBuf writeInt(int value);
    public abstract ByteBuf writeLong(long value);
    
    // ===== 索引操作 =====
    
    public abstract int readerIndex();
    public abstract ByteBuf readerIndex(int readerIndex);
    public abstract int writerIndex();
    public abstract ByteBuf writerIndex(int writerIndex);
    public abstract int readableBytes();  // writerIndex - readerIndex
    public abstract int writableBytes();  // capacity - writerIndex
    
    // ===== 标记与重置 =====
    
    public abstract ByteBuf markReaderIndex();
    public abstract ByteBuf resetReaderIndex();
    public abstract ByteBuf markWriterIndex();
    public abstract ByteBuf resetWriterIndex();
    
    // ===== 容量操作 =====
    
    public abstract int capacity();
    public abstract ByteBuf capacity(int newCapacity);
    public abstract ByteBuf ensureWritable(int minWritableBytes);
    
    // ===== 零拷贝操作 =====
    
    public abstract ByteBuf slice();  // 切片,共享底层数组
    public abstract ByteBuf slice(int index, int length);
    public abstract ByteBuf duplicate();  // 复制,共享底层数组
    public abstract ByteBuf copy();  // 深拷贝
    
    // ===== 组合缓冲区 =====
    
    public abstract ByteBuf retainedSlice();
    public abstract ByteBuf retainedDuplicate();
}

ByteBuf (抽象类)
    ├─ AbstractByteBuf (抽象基类)
    │   ├─ AbstractReferenceCountedByteBuf (引用计数)
    │   │   ├─ PooledByteBuf (池化)
    │   │   │   ├─ PooledHeapByteBuf (池化堆内存)
    │   │   │   ├─ PooledDirectByteBuf (池化直接内存)
    │   │   │   ├─ PooledUnsafeHeapByteBuf (池化不安全堆内存)
    │   │   │   └─ PooledUnsafeDirectByteBuf (池化不安全直接内存)
    │   │   └─ UnpooledByteBuf (非池化)
    │   │       ├─ UnpooledHeapByteBuf (非池化堆内存)
    │   │       └─ UnpooledDirectByteBuf (非池化直接内存)
    │   └─ EmptyByteBuf (空缓冲区)
    └─ CompositeByteBuf (组合缓冲区)

// 1. 使用 Unpooled 工具类创建 (推荐)
ByteBuf heapBuffer = Unpooled.buffer(256);  // 堆内存
ByteBuf directBuffer = Unpooled.directBuffer(256);  // 直接内存
ByteBuf wrappedBuffer = Unpooled.wrappedBuffer(new byte[128]);  // 包装字节数组
ByteBuf copiedBuffer = Unpooled.copiedBuffer("Hello", StandardCharsets.UTF_8);  // 复制字符串

// 2. 使用 ByteBufAllocator 创建 (推荐,支持池化)
ByteBufAllocator allocator = ByteBufAllocator.DEFAULT;
ByteBuf buffer = allocator.buffer(256);  // 自动选择堆或直接内存
ByteBuf heapBuffer = allocator.heapBuffer(256);  // 堆内存
ByteBuf directBuffer = allocator.directBuffer(256);  // 直接内存
ByteBuf ioBuffer = allocator.ioBuffer(256);  // IO 缓冲区 (通常为直接内存)

// 3. 从 Channel 获取 Allocator
ByteBufAllocator alloc = ctx.alloc();
ByteBuf buffer = alloc.buffer();

// 4. 组合多个 ByteBuf (零拷贝)
ByteBuf header = Unpooled.copiedBuffer("Header", StandardCharsets.UTF_8);
ByteBuf body = Unpooled.copiedBuffer("Body", StandardCharsets.UTF_8);
CompositeByteBuf composite = Unpooled.wrappedBuffer(header, body);

// 1. slice() 切片 - 共享底层数组
ByteBuf source = Unpooled.copiedBuffer("Hello World", StandardCharsets.UTF_8);
ByteBuf slice = source.slice(0, 5);  // 切片 "Hello"
// slice 和 source 共享同一个底层数组,没有数据复制

// 2. duplicate() 复制 - 共享底层数组
ByteBuf duplicate = source.duplicate();
// duplicate 和 source 共享底层数组,只是独立的读写索引

// 3. CompositeByteBuf 组合缓冲区 - 多个缓冲区逻辑组合
ByteBuf header = Unpooled.copiedBuffer("Header\n", StandardCharsets.UTF_8);
ByteBuf body = Unpooled.copiedBuffer("Body\n", StandardCharsets.UTF_8);
ByteBuf footer = Unpooled.copiedBuffer("Footer", StandardCharsets.UTF_8);

// × 传统方式:复制数据到新缓冲区
ByteBuf traditional = Unpooled.buffer(header.readableBytes() + body.readableBytes() + footer.readableBytes());
traditional.writeBytes(header);
traditional.writeBytes(body);
traditional.writeBytes(footer);

// √ 零拷贝方式:逻辑组合,无内存复制
CompositeByteBuf composite = Unpooled.wrappedBuffer(header, body, footer);

// 4. FileRegion 文件传输 - 操作系统零拷贝
File file = new File("large-file.txt");
FileInputStream in = new FileInputStream(file);
FileRegion region = new DefaultFileRegion(in.getChannel(), 0, file.length());
ctx.writeAndFlush(region);  // 使用 sendfile 系统调用,零拷贝

// 5. transferTo 实现文件传输
public void transferFile(ChannelHandlerContext ctx, File file) throws IOException {
    RandomAccessFile raf = new RandomAccessFile(file, "r");
    long length = raf.length();
    DefaultFileRegion region = new DefaultFileRegion(raf.getChannel(), 0, length);
    ctx.writeAndFlush(region).addListener(future -> {
        raf.close();
        if (!future.isSuccess()) {
            future.cause().printStackTrace();
        }
    });
}

Arena (内存区域)
    ├─ HeapArena (堆内存区域)
    │   ├─ SmallSubpagePools (小页内存池)
    │   └─ NormalPool (普通内存池)
    └─ DirectArena (直接内存区域)
        ├─ SmallSubpagePools (小页内存池)
        └─ NormalPool (普通内存池)

PoolThreadCache (线程本地缓存)
    ├─ HeapArena (绑定到当前线程的 Arena)
    └─ DirectArena (绑定到当前线程的 Arena)

// 测试代码
public class ByteBufPerformanceTest {
    private static final int COUNT = 10000000;
    
    public static void main(String[] args) {
        // 非池化测试
        long start = System.nanoTime();
        for (int i = 0; i < COUNT; i++) {
            ByteBuf buf = Unpooled.buffer(1024);
            buf.release();
        }
        long unpooledTime = System.nanoTime() - start;
        
        // 池化测试
        ByteBufAllocator pooledAllocator = new PooledByteBufAllocator(true);
        start = System.nanoTime();
        for (int i = 0; i < COUNT; i++) {
            ByteBuf buf = pooledAllocator.buffer(1024);
            buf.release();
        }
        long pooledTime = System.nanoTime() - start;
        
        System.out.println("Unpooled: " + unpooledTime / 1_000_000 + " ms");
        System.out.println("Pooled: " + pooledTime / 1_000_000 + " ms");
        System.out.println("Pooled is " + (unpooledTime * 1.0 / pooledTime) + "x faster");
    }
}

// 结果示例:
// Unpooled: 2845 ms
// Pooled: 856 ms
// Pooled is 3.32x faster

// 启动参数配置
// -Dio.netty.allocator.type=pooled  # 使用池化分配器 (默认)
// -Dio.netty.allocator.type=unpooled  # 使用非池化分配器

// 代码配置
Bootstrap bootstrap = new Bootstrap();
bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
serverBootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

// 自定义池化配置
PooledByteBufAllocator allocator = new PooledByteBufAllocator(
    true,  // preferDirect
    0,     // nHeapArena
    0,     // nDirectArena
    8192,  // pageSize
    11,    // maxOrder
    0,     // tinyCacheSize
    0,     // smallCacheSize
    0      // normalCacheSize
);

public interface ReferenceCounted {
    // 获取引用计数
    int refCnt();
    
    // 增加引用计数
    ReferenceCounted retain();
    ReferenceCounted retain(int increment);
    
    // 记录引用计数 (用于调试)
    boolean touch();
    boolean touch(Object hint);
    
    // 减少引用计数,如果计数为 0 则释放资源
    boolean release();
    boolean release(int decrement);
}

// 方式 1: 使用 SimpleChannelInboundHandler (推荐)
public class SimpleHandler extends SimpleChannelInboundHandler<ByteBuf> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
        // SimpleChannelInboundHandler 会自动释放 msg
        String data = msg.toString(StandardCharsets.UTF_8);
        System.out.println("Received: " + data);
    }
}

// 方式 2: 使用 ChannelInboundHandlerAdapter 手动释放
public class ManualHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            ByteBuf buf = (ByteBuf) msg;
            String data = buf.toString(StandardCharsets.UTF_8);
            System.out.println("Received: " + data);
        } finally {
            // 手动释放,必须调用
            ReferenceCountUtil.release(msg);
        }
    }
}

// 方式 3: 使用 try-with-resources (ByteBuf 实现了 ReferenceCounted)
public class TryWithResourcesHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf buf = (ByteBuf) msg;
        try {
            String data = buf.toString(StandardCharsets.UTF_8);
            System.out.println("Received: " + data);
        } finally {
            buf.release();
        }
    }
}

// 方式 4: 传递给下一个 Handler
public class ForwardHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 增加引用计数
        ReferenceCountUtil.retain(msg);
        
        // 传递给下一个 Handler
        ctx.fireChannelRead(msg);
        
        // 释放当前引用
        ReferenceCountUtil.release(msg);
    }
}

// 启用内存泄漏检测 (开发环境)
// -Dio.netty.leakDetection.level=SIMPLE  # 简单检测 (默认)
// -Dio.netty.leakDetection.level=ADVANCED  # 高级检测
// -Dio.netty.leakDetection.level=PARANOID  # 偏执检测 (最详细,性能影响大)

// ResourceLeakDetector 源码关键逻辑
public enum Level {
    DISABLED,   // 禁用泄漏检测
    SIMPLE,     // 简单检测,1/128 采样率
    ADVANCED,   // 高级检测,1/128 采样率,报告泄漏对象的访问记录
    PARANOID    // 偏执检测,100% 采样,报告所有对象的访问记录
}

// 检测到泄漏时的日志示例
// LEAK: ByteBuf.release() was not called before it's garbage-collected. 
// Recent access records: 
// #1: io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:286)
// #2: io.netty.channel.DefaultChannelPipeline.touch(DefaultChannelPipeline.java:116)

ByteToMessageDecoder (字节到消息解码器)
    ├─ FixedLengthFrameDecoder (固定长度拆包)
    ├─ LineBasedFrameDecoder (换行符拆包)
    ├─ DelimiterBasedFrameDecoder (分隔符拆包)
    ├─ LengthFieldBasedFrameDecoder (长度字段拆包)
    ├─ JsonObjectDecoder (JSON 对象解码)
    ├─ Base64Decoder (Base64 解码)
    └─ ReplayingDecoder<S> (可重放解码器)

MessageToMessageDecoder<I> (消息到消息解码器)
    ├─ StringDecoder (字符串解码)
    ├─ JsonObjectDecoder (JSON 解码)
    └─ 自定义解码器

MessageToByteEncoder<I> (消息到字节编码器)
    ├─ LengthFieldPrepender (长度字段编码)
    ├─ Base64Encoder (Base64 编码)
    └─ 自定义编码器

MessageToMessageEncoder<I> (消息到消息编码器)
    ├─ StringEncoder (字符串编码)
    └─ 自定义编码器

+----------------+----------------------+
| 4字节消息长度   |     真实消息内容       |
+----------------+----------------------+

// 每条消息固定 20 字节
pipeline.addLast(new FixedLengthFrameDecoder(20));

// 适用场景:
// - 消息长度固定
// - 简单的二进制协议
// - 空间利用率低,不推荐

// 使用换行符 \n 或 \r\n 作为消息分隔符
pipeline.addLast(new LineBasedFrameDecoder(1024));  // 最大帧长度 1024
pipeline.addLast(new StringDecoder());

// 适用场景:
// - 文本协议
// - 简单的命令行协议
// - HTTP 头部解析

// 使用自定义分隔符
ByteBuf delimiter = Unpooled.copiedBuffer("$$$", StandardCharsets.UTF_8);
pipeline.addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
pipeline.addLast(new StringDecoder());

// 多个分隔符
ByteBuf delimiter1 = Unpooled.copiedBuffer("\r\n", StandardCharsets.UTF_8);
ByteBuf delimiter2 = Unpooled.copiedBuffer("\n", StandardCharsets.UTF_8);
pipeline.addLast(new DelimiterBasedFrameDecoder(1024, delimiter1, delimiter2));

// 适用场景:
// - 自定义文本协议
// - 需要特殊分隔符的场景

public LengthFieldBasedFrameDecoder(
    int maxFrameLength,         // 最大帧长度
    int lengthFieldOffset,      // 长度字段偏移量
    int lengthFieldLength,      // 长度字段长度 (1/2/3/4/8 字节)
    int lengthAdjustment,       // 长度调整值
    int initialBytesToStrip     // 跳过的字节数
)

示例 1: 最简单的情况
协议格式: [长度(4字节)][消息体]
数据:     [00 00 00 0C][48 65 6C 6C 6F 20 57 6F 72 6C 64 21]
          (长度=12)    (Hello World!)

参数:
  maxFrameLength = 1024
  lengthFieldOffset = 0      // 长度字段从第 0 字节开始
  lengthFieldLength = 4      // 长度字段占 4 字节
  lengthAdjustment = 0       // 无需调整
  initialBytesToStrip = 4    // 跳过长度字段

解码后: [48 65 6C 6C 6F 20 57 6F 72 6C 64 21] (Hello World!)

示例 2: 带版本号的协议
协议格式: [版本号(2字节)][长度(4字节)][消息体]
数据:     [00 01][00 00 00 0C][48 65 6C 6C 6F 20 57 6F 72 6C 64 21]
          (v1)  (长度=12)    (Hello World!)

参数:
  maxFrameLength = 1024
  lengthFieldOffset = 2      // 长度字段从第 2 字节开始
  lengthFieldLength = 4      // 长度字段占 4 字节
  lengthAdjustment = 0       // 无需调整
  initialBytesToStrip = 6    // 跳过版本号和长度字段

解码后: [48 65 6C 6C 6F 20 57 6F 72 6C 64 21] (Hello World!)

示例 3: 长度字段包含自身长度
协议格式: [总长度(4字节)][消息体]
数据:     [00 00 00 10][48 65 6C 6C 6F 20 57 6F 72 6C 64 21]
          (长度=16)     (Hello World!)
          注: 16 = 4(长度字段) + 12(消息体)

参数:
  maxFrameLength = 1024
  lengthFieldOffset = 0      // 长度字段从第 0 字节开始
  lengthFieldLength = 4      // 长度字段占 4 字节
  lengthAdjustment = -4      // 减去长度字段本身
  initialBytesToStrip = 4    // 跳过长度字段

解码后: [48 65 6C 6C 6F 20 57 6F 72 6C 64 21] (Hello World!)

示例 4: 带魔数和版本号
协议格式: [魔数(4字节)][版本号(2字节)][长度(4字节)][消息体]
数据:     [CA FE BA BE][00 01][00 00 00 0C][48 65 6C 6C 6F ...]

参数:
  maxFrameLength = 1024
  lengthFieldOffset = 6      // 长度字段从第 6 字节开始
  lengthFieldLength = 4      // 长度字段占 4 字节
  lengthAdjustment = 0       // 无需调整
  initialBytesToStrip = 10   // 跳过魔数、版本号、长度字段

解码后: [48 65 6C 6C 6F 20 57 6F 72 6C 64 21] (Hello World!)

// 协议格式: [长度(4字节)][消息体]
pipeline.addLast(new LengthFieldBasedFrameDecoder(
    1024,   // 最大帧长度 1024 字节
    0,      // 长度字段从第 0 字节开始
    4,      // 长度字段占 4 字节
    0,      // 无需调整
    4       // 跳过长度字段
));
pipeline.addLast(new StringDecoder(StandardCharsets.UTF_8));
pipeline.addLast(new StringEncoder(StandardCharsets.UTF_8));
pipeline.addLast(new LengthFieldPrepender(4));  // 出站时自动添加长度字段

// 自定义消息格式: [魔数(4字节)][版本号(2字节)][长度(4字节)][消息体]
public class CustomMessage {
    private int magic;       // 魔数
    private short version;   // 版本号
    private byte[] body;     // 消息体
}

// 编码器
public class CustomEncoder extends MessageToByteEncoder<CustomMessage> {
    @Override
    protected void encode(ChannelHandlerContext ctx, CustomMessage msg, ByteBuf out) {
        out.writeInt(msg.getMagic());
        out.writeShort(msg.getVersion());
        out.writeInt(msg.getBody().length);
        out.writeBytes(msg.getBody());
    }
}

// 解码器
public class CustomDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // 检查是否可读至少 10 字节 (魔数 + 版本号 + 长度)
        if (in.readableBytes() < 10) {
            return;
        }
        
        // 标记读索引
        in.markReaderIndex();
        
        // 读取魔数
        int magic = in.readInt();
        if (magic != 0xCAFEBAFE) {
            ctx.close();
            return;
        }
        
        // 读取版本号
        short version = in.readShort();
        
        // 读取长度
        int length = in.readInt();
        if (length < 0 || length > 1024) {
            ctx.close();
            return;
        }
        
        // 检查是否可读完整的消息体
        if (in.readableBytes() < length) {
            in.resetReaderIndex();
            return;
        }
        
        // 读取消息体
        byte[] body = new byte[length];
        in.readBytes(body);
        
        // 创建消息对象
        CustomMessage message = new CustomMessage(magic, version, body);
        out.add(message);
    }
}

public class CustomReplayingDecoder extends ReplayingDecoder<Void> {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // ReplayingDecoder 会自动检查可读字节数
        // 如果不够会自动等待,无需手动检查
        
        int magic = in.readInt();    // 如果可读字节不够,会等待
        short version = in.readShort();
        int length = in.readInt();
        
        byte[] body = new byte[length];
        in.readBytes(body);          // 如果可读字节不够,会等待
        
        out.add(new CustomMessage(magic, version, body));
    }
}

// 1. 拆包器 + 解码器组合
pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(1024, 0, 4, 0, 4));
pipeline.addLast("messageDecoder", new CustomDecoder());

// 2. 编码器组合
pipeline.addLast("messageEncoder", new CustomEncoder());

// 3. 使用 @Sharable 共享编解码器 (注意线程安全)
@Sharable
public class SharedStringDecoder extends MessageToMessageDecoder<ByteBuf> {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) {
        out.add(msg.toString(StandardCharsets.UTF_8));
    }
}

// 4. 异常处理
pipeline.addLast(new ChannelInboundHandlerAdapter() {
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        if (cause instanceof TooLongFrameException) {
            // 帧长度超过限制
            ctx.close();
        } else if (cause instanceof CorruptedFrameException) {
            // 帧数据损坏
            ctx.close();
        }
    }
});

package com.example.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.nio.charset.StandardCharsets;

public class NettyEchoServer {

    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) {
                            channel.pipeline()
                                    // 入站：先按长度字段拆包
                                    .addLast(new LengthFieldBasedFrameDecoder(1024, 0, 4, 0, 4))
                                    // 入站：把 ByteBuf 解码成字符串
                                    .addLast(new StringDecoder(StandardCharsets.UTF_8))
                                    // 出站：在响应前补 4 字节长度字段
                                    .addLast(new LengthFieldPrepender(4))
                                    // 出站：把字符串编码成字节流
                                    .addLast(new StringEncoder(StandardCharsets.UTF_8))
                                    // 业务处理
                                    .addLast(new SimpleChannelInboundHandler<String>() {
                                        @Override
                                        public void channelActive(ChannelHandlerContext ctx) {
                                            System.out.println("客户端已连接：" + ctx.channel().remoteAddress());
                                        }

                                        @Override
                                        protected void channelRead0(ChannelHandlerContext ctx, String message) {
                                            System.out.println("收到消息：" + message);
                                            ctx.writeAndFlush("服务端响应：" + message);
                                        }

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

            ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();
            System.out.println("Netty 服务端已启动，端口：8080");
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully().sync();
            workerGroup.shutdownGracefully().sync();
        }
    }
}

// 服务端配置
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap
    .group(bossGroup, workerGroup)
    .channel(NioServerSocketChannel.class)
    
    // ===== 服务端 Socket 参数 =====
    .option(ChannelOption.SO_BACKLOG, 1024)  // 等待连接队列最大长度
    .option(ChannelOption.SO_REUSEADDR, true)  // 允许重用本地地址
    
    // ===== 客户端 Socket 参数 =====
    .childOption(ChannelOption.SO_KEEPALIVE, true)  // 启用心跳检测
    .childOption(ChannelOption.TCP_NODELAY, true)  // 禁用 Nagle 算法
    .childOption(ChannelOption.SO_SNDBUF, 32 * 1024)  // 发送缓冲区 32KB
    .childOption(ChannelOption.SO_RCVBUF, 32 * 1024)  // 接收缓冲区 32KB
    
    // ===== Netty 特有参数 =====
    .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)  // 池化分配器
    .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, 
        new WriteBufferWaterMark(32 * 1024, 64 * 1024))  // 写缓冲区水位
    .childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);  // 连接超时 5 秒

// TCP 三次握手过程中有两个队列:
// 1. SYN 队列: 存放已收到 SYN 包但未完成三次握手的连接
// 2. Accept 队列: 存放已完成三次握手但未被应用 accept() 的连接

// SO_BACKLOG 在 Linux 中定义了 Accept 队列的最大长度
// 如果队列满了,客户端会收到 ECONNREFUSED 错误

// 推荐配置:
.option(ChannelOption.SO_BACKLOG, 1024)  // 高并发场景

// Linux 默认值:
// /proc/sys/net/ipv4/tcp_max_syn_backlog  (SYN 队列长度,默认 128)
// /proc/sys/net/core/somaxconn  (Accept 队列最大长度,默认 128)

// Nagle 算法:
// 将多个小包合并成一个大包发送,减少网络负载
// 但会增加延迟,不适合实时性要求高的场景

// 禁用 Nagle 算法后:
// 每次写入都会立即发送,减少延迟

// 推荐配置:
.childOption(ChannelOption.TCP_NODELAY, true)  // 低延迟场景 (推荐)
.childOption(ChannelOption.TCP_NODELAY, false)  // 大数据传输场景

// Netty 写缓冲区水位控制
// 当写缓冲区中的字节数超过高水位时,channel.isWritable() 返回 false
// 当写缓冲区中的字节数低于低水位时,channel.isWritable() 返回 true

WriteBufferWaterMark waterMark = new WriteBufferWaterMark(
    32 * 1024,  // 低水位 32KB
    64 * 1024   // 高水位 64KB
);
.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, waterMark);

// 使用示例
if (ctx.channel().isWritable()) {
    ctx.writeAndFlush(message);
} else {
    // 写缓冲区满,等待或丢弃消息
    System.out.println("Write buffer is full");
}

// 缓冲区大小影响吞吐量和延迟
// 太小: 频繁阻塞,降低吞吐量
// 太大: 占用过多内存,增加 GC 压力

// 推荐配置 (根据业务场景调整):
.childOption(ChannelOption.SO_SNDBUF, 32 * 1024)  // 发送缓冲区 32KB
.childOption(ChannelOption.SO_RCVBUF, 32 * 1024)  // 接收缓冲区 32KB

// 高吞吐量场景:
.childOption(ChannelOption.SO_SNDBUF, 256 * 1024)  // 发送缓冲区 256KB
.childOption(ChannelOption.SO_RCVBUF, 256 * 1024)  // 接收缓冲区 256KB

// 查看系统默认值:
// Linux: cat /proc/sys/net/ipv4/tcp_wmem
//        cat /proc/sys/net/ipv4/tcp_rmem

// 启用 TCP 层面的心跳检测
// 如果连接空闲超过一定时间,TCP 会发送心跳包检测连接是否存活

.childOption(ChannelOption.SO_KEEPALIVE, true)

// Linux 默认参数:
// /proc/sys/net/ipv4/tcp_keepalive_time = 7200 (秒,2 小时)
// /proc/sys/net/ipv4/tcp_keepalive_intvl = 75 (秒,心跳间隔)
// /proc/sys/net/ipv4/tcp_keepalive_probes = 9 (失败次数)

// 注意: Netty 的 IdleStateHandler 是应用层心跳,更灵活

// 检测是否支持 epoll
if (Epoll.isAvailable()) {
    // 使用 epoll
    EventLoopGroup bossGroup = new EpollEventLoopGroup(1);
    EventLoopGroup workerGroup = new EpollEventLoopGroup();
    
    ServerBootstrap serverBootstrap = new ServerBootstrap();
    serverBootstrap
        .group(bossGroup, workerGroup)
        .channel(EpollServerSocketChannel.class)
        .option(ChannelOption.SO_BACKLOG, 1024)
        .childOption(ChannelOption.EPOLL_MODE, EpollMode.LEVEL_TRIGGERED);
} else {
    // 回退到 NIO
    EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    
    ServerBootstrap serverBootstrap = new ServerBootstrap();
    serverBootstrap
        .group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class);
}

// 水平触发 (Level Triggered,默认)
.childOption(ChannelOption.EPOLL_MODE, EpollMode.LEVEL_TRIGGERED);
// 特点: 只要文件描述符就绪,就会一直触发事件
// 适用: 大多数场景

// 边缘触发 (Edge Triggered)
.childOption(ChannelOption.EPOLL_MODE, EpollMode.EDGE_TRIGGERED);
// 特点: 只在文件描述符状态变化时触发事件
// 适用: 高性能场景,需要一次性读完所有数据

// 使用池化 ByteBuf
.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

// 对象复用
public class Request {
    private static final ObjectPool<Request> RECYCLER = ObjectPool.newPool(
        handle -> new Request(handle)
    );
    
    public static Request newInstance() {
        return RECYCLER.get();
    }
    
    public void recycle() {
        handle.recycle(this);
    }
}

// 避免在 Handler 中创建大量临时对象
public class OptimizedHandler extends ChannelInboundHandlerAdapter {
    private final ByteBuf buffer = Unpooled.buffer(1024);  // 复用缓冲区
    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 复用 buffer
        buffer.clear();
        // ... 处理逻辑
    }
}

// 使用 @Sharable 共享无状态的 Handler
@Sharable
public class StatelessHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 无状态处理,线程安全
        ctx.fireChannelRead(msg);
    }
}

// 注册为单例
public class ServerInitializer extends ChannelInitializer<SocketChannel> {
    private static final StatelessHandler SHARED_HANDLER = new StatelessHandler();
    
    @Override
    protected void initChannel(SocketChannel ch) {
        ch.pipeline().addLast(SHARED_HANDLER);  // 共享同一个实例
    }
}

public class AsyncBusinessHandler extends SimpleChannelInboundHandler<String> {
    private static final ExecutorService BUSINESS_POOL = Executors.newFixedThreadPool(16);
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String request) {
        // 提交到业务线程池
        BUSINESS_POOL.submit(() -> {
            String result = processBusiness(request);
            
            // 回到 EventLoop 发送响应
            ctx.channel().eventLoop().execute(() -> {
                ctx.writeAndFlush(result);
            });
        });
    }
    
    private String processBusiness(String request) {
        // 耗时业务逻辑
        return "Processed: " + request;
    }
}

public class BatchWriteHandler extends ChannelInboundHandlerAdapter {
    private final List<String> batch = new ArrayList<>();
    private final int batchSize = 100;
    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        batch.add((String) msg);
        
        if (batch.size() >= batchSize) {
            // 批量写入
            ctx.writeAndFlush(new ArrayList<>(batch));
            batch.clear();
        }
    }
    
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        // 刷新剩余消息
        if (!batch.isEmpty()) {
            ctx.writeAndFlush(new ArrayList<>(batch));
            batch.clear();
        }
        ctx.flush();
    }
}

// 生产环境配置
// -Dio.netty.leakDetection.level=SIMPLE

// 开发环境配置
// -Dio.netty.leakDetection.level=PARANOID

// 代码中检测
ByteBuf buf = ctx.alloc().buffer();
try {
    // ... 使用 buf
} finally {
    int refCnt = buf.refCnt();
    if (refCnt > 0) {
        buf.release(refCnt);  // 释放所有引用
    }
}

// 消息类型
public enum MessageType {
    LOGIN_REQUEST(1),      // 登录请求
    LOGIN_RESPONSE(2),     // 登录响应
    PRIVATE_MESSAGE(3),    // 私聊消息
    GROUP_MESSAGE(4),      // 群聊消息
    HEARTBEAT(5);          // 心跳包
    
    private final int type;
    
    MessageType(int type) {
        this.type = type;
    }
    
    public int getType() {
        return type;
    }
    
    public static MessageType valueOf(int type) {
        for (MessageType mt : values()) {
            if (mt.type == type) {
                return mt;
            }
        }
        return null;
    }
}

// 消息头
public class MessageHeader {
    private int magic;           // 魔数 0xCAFE
    private short version;       // 版本号
    private int type;            // 消息类型
    private long sequenceId;     // 序列号
    private int length;          // 消息体长度
}

// 消息体
public class Message {
    private MessageHeader header;
    private Object body;         // JSON 字符串或二进制数据
}

// 消息编码器
public class MessageEncoder extends MessageToByteEncoder<Message> {
    @Override
    protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) {
        // 1. 写魔数
        out.writeInt(0xCAFE);
        
        // 2. 写版本号
        out.writeShort(1);
        
        // 3. 写消息类型
        out.writeInt(msg.getHeader().getType());
        
        // 4. 写序列号
        out.writeLong(msg.getHeader().getSequenceId());
        
        // 5. 写消息体
        byte[] bodyBytes = serialize(msg.getBody());
        out.writeInt(bodyBytes.length);
        out.writeBytes(bodyBytes);
    }
    
    private byte[] serialize(Object obj) {
        // JSON 序列化
        return JSON.toJSONString(obj).getBytes(StandardCharsets.UTF_8);
    }
}

// 消息解码器
public class MessageDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        // 检查魔数
        if (in.readableBytes() < 22) {  // 最小头部大小
            return;
        }
        
        in.markReaderIndex();
        
        int magic = in.readInt();
        if (magic != 0xCAFE) {
            ctx.close();
            return;
        }
        
        short version = in.readShort();
        int type = in.readInt();
        long sequenceId = in.readLong();
        int length = in.readInt();
        
        if (length < 0 || length > 1024 * 1024) {  // 最大 1MB
            ctx.close();
            return;
        }
        
        if (in.readableBytes() < length) {
            in.resetReaderIndex();
            return;
        }
        
        byte[] bodyBytes = new byte[length];
        in.readBytes(bodyBytes);
        
        MessageHeader header = new MessageHeader(magic, version, type, sequenceId, length);
        Object body = deserialize(bodyBytes, MessageType.valueOf(type));
        
        out.add(new Message(header, body));
    }
    
    private Object deserialize(byte[] bytes, MessageType type) {
        String json = new String(bytes, StandardCharsets.UTF_8);
        switch (type) {
            case LOGIN_REQUEST:
                return JSON.parseObject(json, LoginRequest.class);
            case PRIVATE_MESSAGE:
                return JSON.parseObject(json, PrivateMessage.class);
            case GROUP_MESSAGE:
                return JSON.parseObject(json, GroupMessage.class);
            default:
                return json;
        }
    }
}

public class IMServer {
    private final EventLoopGroup bossGroup;
    private final EventLoopGroup workerGroup;
    private final Map<String, Channel> userChannels = new ConcurrentHashMap<>();
    
    public IMServer() {
        bossGroup = new NioEventLoopGroup(1);
        workerGroup = new NioEventLoopGroup();
    }
    
    public void start(int port) throws InterruptedException {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .option(ChannelOption.SO_BACKLOG, 1024)
            .childOption(ChannelOption.SO_KEEPALIVE, true)
            .childOption(ChannelOption.TCP_NODELAY, true)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    ch.pipeline()
                        // 空闲检测
                        .addLast(new IdleStateHandler(60, 0, 0))
                        // 编解码器
                        .addLast(new MessageDecoder())
                        .addLast(new MessageEncoder())
                        // 业务处理器
                        .addLast(new LoginHandler(userChannels))
                        .addLast(new PrivateMessageHandler(userChannels))
                        .addLast(new GroupMessageHandler(userChannels))
                        .addLast(new HeartbeatHandler())
                        // 异常处理
                        .addLast(new ExceptionHandler());
                }
            });
        
        ChannelFuture future = bootstrap.bind(port).sync();
        System.out.println("IM Server started on port " + port);
        future.channel().closeFuture().sync();
    }
    
    public void shutdown() {
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}

// 登录处理器
public class LoginHandler extends SimpleChannelInboundHandler<LoginRequest> {
    private final Map<String, Channel> userChannels;
    
    public LoginHandler(Map<String, Channel> userChannels) {
        this.userChannels = userChannels;
    }
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, LoginRequest request) {
        String userId = request.getUserId();
        
        // 保存用户连接
        userChannels.put(userId, ctx.channel());
        
        // 绑定用户ID到 Channel
        ctx.channel().attr(AttributeKey.valueOf("userId")).set(userId);
        
        // 发送登录成功响应
        LoginResponse response = new LoginResponse(true, "Login success");
        ctx.writeAndFlush(new Message(
            new MessageHeader(0xCAFE, (short) 1, MessageType.LOGIN_RESPONSE.getType(), 
                System.currentTimeMillis(), 0),
            response
        ));
        
        System.out.println("User " + userId + " logged in");
    }
}

// 私聊处理器
public class PrivateMessageHandler extends SimpleChannelInboundHandler<PrivateMessage> {
    private final Map<String, Channel> userChannels;
    
    public PrivateMessageHandler(Map<String, Channel> userChannels) {
        this.userChannels = userChannels;
    }
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, PrivateMessage msg) {
        String toUserId = msg.getToUserId();
        Channel targetChannel = userChannels.get(toUserId);
        
        if (targetChannel != null && targetChannel.isActive()) {
            // 转发消息
            targetChannel.writeAndFlush(new Message(
                new MessageHeader(0xCAFE, (short) 1, MessageType.PRIVATE_MESSAGE.getType(),
                    System.currentTimeMillis(), 0),
                msg
            ));
        } else {
            // 用户不在线
            ctx.writeAndFlush(new Message(
                new MessageHeader(0xCAFE, (short) 1, MessageType.PRIVATE_MESSAGE.getType(),
                    System.currentTimeMillis(), 0),
                new PrivateMessage("system", msg.getFromUserId(), 
                    "User " + toUserId + " is offline")
            ));
        }
    }
}

public class FileServerHandler extends SimpleChannelInboundHandler<FileUploadRequest> {
    private final String baseDir = "/data/uploads";
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FileUploadRequest request) {
        String fileName = request.getFileName();
        long fileSize = request.getFileSize();
        String fileMd5 = request.getFileMd5();
        
        // 创建临时文件
        Path tempFile = Paths.get(baseDir, fileName + ".tmp");
        
        // 保存文件信息到 Channel 属性
        ctx.channel().attr(AttributeKey.valueOf("tempFile")).set(tempFile);
        ctx.channel().attr(AttributeKey.valueOf("fileMd5")).set(fileMd5);
        ctx.channel().attr(AttributeKey.valueOf("receivedBytes")).set(0L);
        ctx.channel().attr(AttributeKey.valueOf("fileSize")).set(fileSize);
        
        // 发送准备接收响应
        ctx.writeAndFlush(new FileUploadResponse(true, "Ready to receive"));
    }
    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof FileChunk) {
            FileChunk chunk = (FileChunk) msg;
            handleFileChunk(ctx, chunk);
        } else {
            // 交给父类按泛型类型分发到 channelRead0(否则 channelRead0 永远不会被调用)
            super.channelRead(ctx, msg);
        }
    }
    
    private void handleFileChunk(ChannelHandlerContext ctx, FileChunk chunk) {
        Path tempFile = (Path) ctx.channel().attr(AttributeKey.valueOf("tempFile")).get();
        long receivedBytes = (Long) ctx.channel().attr(AttributeKey.valueOf("receivedBytes")).get();
        long fileSize = (Long) ctx.channel().attr(AttributeKey.valueOf("fileSize")).get();
        
        try {
            // 写入文件块
            Files.write(tempFile, chunk.getData(), 
                StandardOpenOption.CREATE, StandardOpenOption.APPEND);
            
            receivedBytes += chunk.getData().length;
            ctx.channel().attr(AttributeKey.valueOf("receivedBytes")).set(receivedBytes);
            
            // 发送进度
            double progress = (double) receivedBytes / fileSize * 100;
            ctx.writeAndFlush(new FileProgress(progress));
            
            // 检查是否接收完成
            if (receivedBytes >= fileSize) {
                verifyAndRenameFile(ctx, tempFile);
            }
        } catch (IOException e) {
            ctx.writeAndFlush(new FileUploadResponse(false, "Upload failed: " + e.getMessage()));
        }
    }
    
    private void verifyAndRenameFile(ChannelHandlerContext ctx, Path tempFile) {
        try {
            // 计算 MD5
            String md5 = calculateMD5(tempFile);
            String expectedMd5 = (String) ctx.channel().attr(AttributeKey.valueOf("fileMd5")).get();
            
            if (md5.equals(expectedMd5)) {
                // 重命名为正式文件
                String fileName = tempFile.getFileName().toString().replace(".tmp", "");
                Path finalFile = tempFile.resolveSibling(fileName);
                Files.move(tempFile, finalFile, StandardCopyOption.REPLACE_EXISTING);
                
                ctx.writeAndFlush(new FileUploadResponse(true, "Upload completed"));
            } else {
                Files.delete(tempFile);
                ctx.writeAndFlush(new FileUploadResponse(false, "MD5 verification failed"));
            }
        } catch (Exception e) {
            ctx.writeAndFlush(new FileUploadResponse(false, "Verification failed: " + e.getMessage()));
        }
    }
    
    private String calculateMD5(Path file) throws IOException {
        MessageDigest md = MessageDigest.getInstance("MD5");
        try (InputStream is = Files.newInputStream(file)) {
            byte[] buffer = new byte[8192];
            int read;
            while ((read = is.read(buffer)) != -1) {
                md.update(buffer, 0, read);
            }
        }
        return Hex.encodeHexString(md.digest());
    }
}

public class FileDownloadHandler extends SimpleChannelInboundHandler<FileDownloadRequest> {
    private final String baseDir = "/data/files";
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FileDownloadRequest request) {
        String fileName = request.getFileName();
        Path file = Paths.get(baseDir, fileName);
        
        if (!Files.exists(file)) {
            ctx.writeAndFlush(new FileDownloadResponse(false, "File not found"));
            return;
        }
        
        try {
            long fileSize = Files.size(file);
            
            // 发送文件信息
            ctx.writeAndFlush(new FileDownloadResponse(true, fileName, fileSize));
            
            // 使用 FileRegion 零拷贝传输
            RandomAccessFile raf = new RandomAccessFile(file.toFile(), "r");
            FileRegion region = new DefaultFileRegion(raf.getChannel(), 0, fileSize);
            
            ctx.writeAndFlush(region).addListener((ChannelFutureListener) future -> {
                raf.close();
                if (!future.isSuccess()) {
                    future.cause().printStackTrace();
                }
            });
        } catch (IOException e) {
            ctx.writeAndFlush(new FileDownloadResponse(false, "Download failed: " + e.getMessage()));
        }
    }
}

public class WebSocketChatServer {
    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline()
                            // HTTP 编解码器
                            .addLast(new HttpServerCodec())
                            .addLast(new HttpObjectAggregator(65536))
                            // WebSocket 协议升级
                            .addLast(new WebSocketServerProtocolHandler("/chat"))
                            // 心跳检测
                            .addLast(new IdleStateHandler(60, 0, 0))
                            // WebSocket 消息处理
                            .addLast(new WebSocketChatHandler())
                            .addLast(new WebSocketHeartbeatHandler());
                    }
                });
            
            ChannelFuture future = bootstrap.bind(8080).sync();
            System.out.println("WebSocket Chat Server started on port 8080");
            future.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

// WebSocket 消息处理器
public class WebSocketChatHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    private static final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        // 新用户加入
        channels.add(ctx.channel());
        channels.writeAndFlush(new TextWebSocketFrame(
            "User " + ctx.channel().id().asShortText() + " joined"
        ));
    }
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) {
        String message = frame.text();
        
        // 广播消息给所有用户
        channels.writeAndFlush(new TextWebSocketFrame(
            "[" + ctx.channel().id().asShortText() + "] " + message
        ));
    }
    
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        // 用户离开
        channels.writeAndFlush(new TextWebSocketFrame(
            "User " + ctx.channel().id().asShortText() + " left"
        ));
    }
    
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent) {
            // 空闲超时,关闭连接
            ctx.close();
        }
    }
}

<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8">
    <title>WebSocket Chat</title>
</head>
<body>
    <div id="messages"></div>
    <input type="text" id="messageInput">
    <button onclick="sendMessage()">Send</button>
    
    <script>
        const ws = new WebSocket('ws://localhost:8080/chat');
        const messagesDiv = document.getElementById('messages');
        const messageInput = document.getElementById('messageInput');
        
        ws.onopen = () => {
            addMessage('Connected to server');
        };
        
        ws.onmessage = (event) => {
            addMessage(event.data);
        };
        
        ws.onclose = () => {
            addMessage('Disconnected from server');
        };
        
        function sendMessage() {
            const message = messageInput.value;
            if (message) {
                ws.send(message);
                messageInput.value = '';
            }
        }
        
        function addMessage(message) {
            const div = document.createElement('div');
            div.textContent = message;
            messagesDiv.appendChild(div);
        }
        
        // 支持回车发送
        messageInput.addEventListener('keypress', (e) => {
            if (e.key === 'Enter') {
                sendMessage();
            }
        });
    </script>
</body>
</html>

// × 错误示例
public class LeakyHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf buf = (ByteBuf) msg;
        String data = buf.toString(StandardCharsets.UTF_8);
        // 忘记释放 buf
        process(data);
    }
}

// √ 方案 1: 使用 SimpleChannelInboundHandler
public class SafeHandler extends SimpleChannelInboundHandler<ByteBuf> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
        String data = msg.toString(StandardCharsets.UTF_8);
        process(data);
        // SimpleChannelInboundHandler 会自动释放 msg
    }
}

// √ 方案 2: 手动释放
public class ManualReleaseHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            ByteBuf buf = (ByteBuf) msg;
            String data = buf.toString(StandardCharsets.UTF_8);
            process(data);
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }
}

// √ 方案 3: 启用内存泄漏检测
// 启动参数: -Dio.netty.leakDetection.level=PARANOID

// × 错误示例
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) {
            // 没有配置连接超时
        }
    });

ChannelFuture future = bootstrap.connect("example.com", 8080);
future.sync();  // 无限等待

// √ 配置连接超时
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
    .channel(NioSocketChannel.class)
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)  // 5 秒超时
    .handler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) {
            // 配置处理器
        }
    });

ChannelFuture future = bootstrap.connect("example.com", 8080);
future.addListener((ChannelFutureListener) f -> {
    if (f.isSuccess()) {
        System.out.println("Connected");
    } else {
        System.out.println("Connect failed: " + f.cause().getMessage());
    }
});

// × 错误示例:最大帧长度设置过小
pipeline.addLast(new LengthFieldBasedFrameDecoder(
    100,   // 最大帧长度 100 字节
    0, 4, 0, 4
));

// √ 调整最大帧长度
pipeline.addLast(new LengthFieldBasedFrameDecoder(
    1024 * 1024,   // 最大帧长度 1MB
    0, 4, 0, 4
));

// √ 自定义异常处理
public class FrameExceptionHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        if (cause instanceof TooLongFrameException) {
            // 记录日志
            logger.warn("Frame too long from {}", ctx.channel().remoteAddress());
            // 关闭连接
            ctx.close();
        } else {
            ctx.fireExceptionCaught(cause);
        }
    }
}

// × 错误示例:在 EventLoop 中执行耗时操作
public class BlockingHandler extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        // 阻塞 EventLoop 线程
        try {
            Thread.sleep(5000);  // 模拟耗时操作
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        ctx.writeAndFlush("Response");
    }
}

// √ 方案 1: 使用业务线程池
public class AsyncHandler extends SimpleChannelInboundHandler<String> {
    private static final ExecutorService BUSINESS_POOL = Executors.newFixedThreadPool(16);
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        BUSINESS_POOL.submit(() -> {
            // 耗时操作
            String result = processBusiness(msg);
            
            // 回到 EventLoop 发送响应
            ctx.channel().eventLoop().execute(() -> {
                ctx.writeAndFlush(result);
            });
        });
    }
}

// √ 方案 2: 使用 DefaultEventExecutorGroup
EventExecutorGroup businessGroup = new DefaultEventExecutorGroup(16);

pipeline.addLast(businessGroup, "businessHandler", new BusinessHandler());

// √ 调整系统参数
// Linux:
// echo 65535 > /proc/sys/net/core/somaxconn
// echo 65535 > /proc/sys/net/ipv4/tcp_max_syn_backlog
// ulimit -n 65535

// √ 调整 Netty 参数
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
    .channel(NioServerSocketChannel.class)
    .option(ChannelOption.SO_BACKLOG, 65535)  // 加大队列
    .option(ChannelOption.SO_REUSEADDR, true);  // 允许重用端口

// 服务端
pipeline.addLast(new IdleStateHandler(60, 0, 0));  // 60秒读空闲
pipeline.addLast(new HeartbeatHandler());

public class HeartbeatHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                // 读空闲超时,关闭连接
                ctx.close();
            }
        }
    }
}

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;

// pipeline 中增加：
// .addLast(new IdleStateHandler(60, 0, 0))
// .addLast(new HeartbeatHandler())

public class HeartbeatHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent event && event.state() == IdleState.READER_IDLE) {
            System.out.println("读空闲超时，关闭连接：" + ctx.channel().remoteAddress());
            ctx.close();
            return;
        }
        ctx.fireUserEventTriggered(evt);
    }
}