# Netty 核心原理

## 问题

1. Netty 的核心组件有哪些？
2. 什么是 EventLoop？线程模型是怎样的？
3. 什么是 ByteBuf？与 Java NIO 的 ByteBuffer 有什么区别？
4. Netty 的零拷贝是如何实现的？
5. 什么是 ChannelPipeline？ChannelHandler 是如何工作的？
6. Netty 如何解决 TCP 粘包/拆包问题？
7. Netty 的心跳机制是如何实现的？

---

## 标准答案

### 1. Netty 核心组件

#### **核心组件架构**

```
┌─────────────────────────────────────────────┐
│              Netty 架构                      │
├─────────────────────────────────────────────┤
│  Channel                                    │
│  ├─ EventLoop (线程模型)                    │
│  ├─ ChannelPipeline (处理器链)              │
│  │   └─ ChannelHandler (业务处理器)         │
│  └─ ByteBuf (内存管理)                      │
├─────────────────────────────────────────────┤
│  Bootstrap (启动类)                         │
│  ├─ ServerBootstrap (服务端)                │
│  └─ Bootstrap (客户端)                      │
└─────────────────────────────────────────────┘
```

#### **关键组件说明**

**1. Channel**
```java
// Channel 是 Netty 的网络操作抽象类
Channel channel = ...;

// 核心方法
channel.writeAndFlush(msg);  // 写数据并刷新
channel.close();              // 关闭连接
channel.eventLoop();          // 获取 EventLoop
```

**2. EventLoop**
```java
// EventLoop 负责处理 I/O 操作
EventLoopGroup bossGroup = new NioEventLoopGroup(1);      // Accept
EventLoopGroup workerGroup = new NioEventLoopGroup();     // I/O

// bossGroup 只负责监听并接受连接
// workerGroup 负责 I/O 读写
```

**3. ChannelPipeline**
```java
// Pipeline 是处理器链
ChannelPipeline pipeline = channel.pipeline();

// 添加处理器
pipeline.addLast("decoder", new Decoder());
pipeline.addLast("encoder", new Encoder());
pipeline.addLast("handler", new BusinessHandler());
```

**4. ByteBuf**
```java
// ByteBuf 是 Netty 的字节容器
ByteBuf buffer = Unpooled.buffer(1024);

// 写入数据
buffer.writeInt(123);
buffer.writeBytes("Hello".getBytes());

// 读取数据
int value = buffer.readInt();
```

---

### 2. EventLoop 与线程模型

#### **Reactor 线程模型**

Netty 采用 **Reactor 模式**，支持三种实现：

```
1. Reactor 单线程模型
   ┌──────────────┐
   │  Reactor     │
   │  (单线程)    │
   │              │
   │  - Accept    │
   │  - Read      │
   │  - Decode    │
   │  - Process   │
   │  - Encode    │
   │  - Send      │
   └──────────────┘

2. Reactor 多线程模型
   ┌──────────┐     ┌──────────────────┐
   │ Reactor  │────>│ Worker Threads   │
   │ (主线程) │     │ (线程池)          │
   │          │     │                  │
   │ - Accept │     │ - Read/Write     │
   │          │     │ - Decode/Encode  │
   │          │     │ - Process        │
   └──────────┘     └──────────────────┘

3. 主从 Reactor 多线程模型（Netty 默认）
   ┌──────────┐     ┌──────────────────┐
   │ Main     │     │ Sub Reactors     │
   │ Reactor  │────>│ (线程池)          │
   │          │     │                  │
   │ - Accept │     │ - Read/Write     │
   │          │     │ - Decode/Encode  │
   │          │     │ - Process        │
   └──────────┘     └──────────────────┘
```

#### **EventLoop 工作原理**

```java
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();

// 每个 EventLoop:
// 1. 维护一个 Selector
// 2. 维护一个任务队列
// 3. 维护一个线程

// 工作流程:
// - 线程启动后，无限循环执行:
//   1. select() - 查询就绪的 Channel
//   2. processSelectedKeys() - 处理 I/O 事件
//   3. runAllTasks() - 执行任务队列中的任务
```

#### **任务调度**

```java
Channel channel = ...;

// 在 EventLoop 线程中执行任务
channel.eventLoop().execute(() -> {
    System.out.println("Task in EventLoop thread");
});

// 定时任务
channel.eventLoop().schedule(() -> {
    System.out.println("Delayed task");
}, 1, TimeUnit.SECONDS);
```

---

### 3. ByteBuf vs ByteBuffer

#### **对比表**

| 特性 | Java NIO ByteBuffer | Netty ByteBuf |
|------|---------------------|---------------|
| **长度** | 固定长度，扩容复杂 | 动态扩容 |
| **读写模式** | flip() 切换 | 独立的读写索引 |
| **引用计数** | 不支持 | 支持（池化） |
| **零拷贝** | 不支持 | 支持（Composite） |
| **性能** | 较低 | 高（对象池优化） |

#### **ByteBuf 三个重要指针**

```
ByteBuf 结构:
┌───────┬──────────┬──────────┬─────────┐
│  0    │ readerIndex│ writerIndex│ capacity│
│       │          │          │         │
│ discard│  readable│ writable │         │
│  bytes │   bytes  │  bytes   │         │
└───────┴──────────┴──────────┴─────────┘

操作:
- mark() / reset(): 标记和重置
- readerIndex(): 读指针
- writerIndex(): 写指针
- capacity(): 容量
- maxCapacity(): 最大容量
```

#### **使用示例**

```java
// 创建 ByteBuf
ByteBuf buffer = Unpooled.buffer(1024);

// 写入数据
buffer.writeBytes("Hello".getBytes());
buffer.writeInt(123);

// 读取数据（不移动读指针）
byte[] data = new byte[5];
buffer.getBytes(buffer.readerIndex(), data);

// 读取数据（移动读指针）
buffer.readBytes(5);

// 引用计数（池化）
buffer.retain();  // 引用计数 +1
buffer.release(); // 引用计数 -1
```

---

### 4. 零拷贝实现

#### **零拷贝的三种实现**

**1. CompositeByteBuf（组合缓冲区）**

```java
// 将多个 ByteBuf 组合成一个逻辑上的 ByteBuf
ByteBuf header = Unpooled.buffer(10);
ByteBuf body = Unpooled.buffer(100);

// 零拷贝组合
CompositeByteBuf message = Unpooled.wrappedBuffer(header, body);

// 物理上不复制数据，只是逻辑组合
```

**2. wrappedBuffer（包装）**

```java
// 包装字节数组，不复制
byte[] bytes = "Hello Netty".getBytes();
ByteBuf buffer = Unpooled.wrappedBuffer(bytes);

// 底层数据共享，无拷贝
```

**3. slice（切片）**

```java
ByteBuf buffer = Unpooled.buffer(100);

// 切片，共享底层内存
ByteBuf sliced = buffer.slice(0, 50);

// 修改会互相影响
sliced.setByte(0, (byte) 1);
```

**4. FileChannel.transferTo（文件传输）**

```java
// 文件传输零拷贝
FileChannel sourceChannel = ...;
FileChannel destChannel = ...;

// 直接在内核空间传输，不经过用户空间
sourceChannel.transferTo(position, count, destChannel);
```

---

### 5. ChannelPipeline 与 ChannelHandler

#### **Pipeline 工作流程**

```
入站事件 (Inbound):
     │
     ▼
┌─────────────────────────────────────┐
│ ChannelPipeline                      │
│                                     │
│  ┌─────────┐  ┌─────────┐  ┌──────┐│
│  │Decoder1 │→ │Decoder2 │→ │Handler││
│  └─────────┘  └─────────┘  └──────┘│
│      ▲            ▲             ▲  │
│      └────────────┴─────────────┘  │
│              (数据流)               │
└─────────────────────────────────────┘

出站事件 (Outbound):
     │
     ▼
┌─────────────────────────────────────┐
│ ChannelPipeline                      │
│                                     │
│  ┌──────┐  ┌─────────┐  ┌─────────┐│
│  │Handler│→ │Encoder1 │→ │Encoder2 ││
│  └──────┘  └─────────┘  └─────────┘│
│      ▲            ▲             ▲  │
│      └────────────┴─────────────┘  │
│              (数据流)               │
└─────────────────────────────────────┘
```

#### **ChannelHandler 接口**

```java
// 入站处理器
@ChannelHandler.Sharable
public class InboundHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 处理读事件
        System.out.println("Received: " + msg);
        ctx.fireChannelRead(msg);  // 传递给下一个 Handler
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        // 连接建立
        System.out.println("Client connected");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        // 连接断开
        System.out.println("Client disconnected");
    }
}

// 出站处理器
public class OutboundHandler extends ChannelOutboundHandlerAdapter {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg,
                      ChannelPromise promise) {
        // 处理写事件
        System.out.println("Sending: " + msg);
        ctx.write(msg, promise);
    }
}
```

#### **Handler 生命周期**

```java
public interface ChannelHandler {

    // 添加到 Pipeline
    void handlerAdded(ChannelHandlerContext ctx);

    // 从 Pipeline 移除
    void handlerRemoved(ChannelHandlerContext ctx);

    // 发生异常
    void exceptionCaught(ChannelHandlerContext ctx, Throwable cause);
}
```

---

### 6. TCP 粘包/拆包解决方案

#### **问题原因**

```
TCP 粘包:
发送方: [包1] [包2] [包3]
接收方: [包1+包2+包3]  // 粘在一起

TCP 拆包:
发送方: [大包]
接收方: [片段1] [片段2]  // 被拆分
```

#### **Netty 提供的解码器**

**1. FixedLengthFrameDecoder（固定长度）**

```java
// 每个消息固定 10 字节
pipeline.addLast("framer", new FixedLengthFrameDecoder(10));
```

**2. LineBasedFrameDecoder（分隔符）**

```java
// 按换行符分割
pipeline.addLast("framer", new LineBasedFrameDecoder(8192));
```

**3. DelimiterBasedFrameDecoder（自定义分隔符）**

```java
// 自定义分隔符
ByteBuf delimiter = Unpooled.copiedBuffer("\t".getBytes());
pipeline.addLast("framer",
    new DelimiterBasedFrameDecoder(8192, delimiter));
```

**4. LengthFieldBasedFrameDecoder（长度字段）**

```java
// 消息格式: [长度(4字节)] [数据]
pipeline.addLast("framer",
    new LengthFieldBasedFrameDecoder(
        8192,        // maxFrameLength
        0,           // lengthFieldOffset
        4,           // lengthFieldLength
        0,           // lengthAdjustment
        0            // initialBytesToStrip
    ));
```

**5. 自定义解码器**

```java
public class CustomDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx,
                          ByteBuf in, List<Object> out) {
        // 检查是否有足够数据
        if (in.readableBytes() < 4) {
            return;
        }

        // 标记读指针
        in.markReaderIndex();

        // 读取长度
        int length = in.readInt();

        // 检查数据是否完整
        if (in.readableBytes() < length) {
            in.resetReaderIndex();
            return;
        }

        // 读取数据
        byte[] data = new byte[length];
        in.readBytes(data);

        out.add(data);
    }
}
```

---

### 7. 心跳机制

#### **IdleStateHandler 实现**

```java
// 读写空闲检测
pipeline.addLast("idleStateHandler",
    new IdleStateHandler(
        30,        // 读空闲时间（秒）
        0,         // 写空闲时间
        0          // 读写空闲时间
    ));

// 心跳处理器
pipeline.addLast("heartbeatHandler", new HeartbeatHandler());
```

#### **心跳处理器**

```java
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) {
                // 读空闲，关闭连接
                System.out.println("Read idle, close connection");
                ctx.close();
            }
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 处理心跳包
        if ("PING".equals(msg)) {
            ctx.writeAndFlush("PONG");
        } else {
            ctx.fireChannelRead(msg);
        }
    }
}
```

---

## P7 加分项

### 深度理解
- **Reactor 模式**：理解单线程、多线程、主从多线程三种模型
- **零拷贝原理**：理解用户空间和内核空间的数据拷贝
- **内存管理**：ByteBuf 的池化、引用计数、内存泄漏

### 实战经验
- **高并发场景**：EventLoopGroup 线程数调优
- **内存调优**：ByteBuf 分配器选择（堆内存 vs 直接内存）
- **性能优化**：避免 Handler 链过长、减少对象创建

### 架构设计
- **协议设计**：自定义协议、编解码器设计
- **异常处理**：优雅关闭、重连机制
- **监控告警**：流量监控、连接数监控、延迟监控

### 常见问题
1. **内存泄漏**：ByteBuf 未释放
2. **线程安全**：共享状态的同步
3. **连接池**：Channel 复用与管理
4. **背压处理**：流量控制与慢消费

---

## 总结

Netty 的核心优势：
1. **高性能**：零拷贝、内存池、Reactor 模式
2. **高可靠性**：心跳机制、重连机制、优雅关闭
3. **易扩展**：Pipeline 机制、丰富的 Handler
4. **成熟稳定**：广泛应用在 Dubbo、gRPC、WebSocket

**最佳实践**：
- 合理设置 EventLoopGroup 线程数
- 使用池化的 ByteBuf
- 及时释放 ByteBuf，避免内存泄漏
- 使用 LengthFieldBasedFrameDecoder 处理粘包
- 添加心跳和重连机制
- 做好监控和日志