ChannelPipeline 与 Handler 事件传播机制

一、ChannelPipeline 是什么？

ChannelPipeline 是 Netty 中负责管理多个 ChannelHandler 的处理链。

每个 Channel 都有自己的 Pipeline。

Pipeline 主要负责：

消息拆包、粘包处理。
字节数据的编码和解码。
业务逻辑处理。
连接建立、关闭及异常事件处理。
出站数据的编码和发送。

例如：

ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
ch.pipeline().addLast(new StringDecoder(StandardCharsets.UTF_8));
ch.pipeline().addLast(new StringEncoder(StandardCharsets.UTF_8));
ch.pipeline().addLast(new MyServerHandler());

Pipeline 中注册了多个 Handler，不代表每个消息都会依次执行所有 Handler。

Netty 会根据事件类型和传播方向选择对应的 Handler。

二、Inbound 和 Outbound 的区别

Netty 的 Handler 主要分为两类：

类型	接口	主要作用
Inbound	ChannelInboundHandler	接收消息、处理连接事件
Outbound	ChannelOutboundHandler	发送消息、处理写入等出站操作

常见组件：

Handler	类型	作用
LineBasedFrameDecoder	Inbound	按换行符拆分消息
StringDecoder	Inbound	ByteBuf 转 String
StringEncoder	Outbound	String 转 ByteBuf
MyServerHandler	Inbound	执行业务逻辑

核心区别：

Inbound：通常从 Head 向 Tail 传播。
Outbound：向 Head 方向传播。
入站事件不会因为经过 OutboundHandler 的位置就执行它。
出站事件也不会执行普通 InboundHandler 的入站方法。

三、消息入站处理流程

Pipeline：

ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
ch.pipeline().addLast(new StringDecoder(StandardCharsets.UTF_8));
ch.pipeline().addLast(new StringEncoder(StandardCharsets.UTF_8));
ch.pipeline().addLast(new MyServerHandler());

假设客户端发送：

Hello\n

入站过程：

客户端
   |
   | 发送 Hello\n
   v
LineBasedFrameDecoder
   |
   | 按换行符拆分消息
   | 得到 ByteBuf("Hello")
   v
StringDecoder
   |
   | ByteBuf 转换为 String
   v
StringEncoder
   |
   | 跳过，因为它是 OutboundHandler
   v
MyServerHandler
   |
   | 收到 String("Hello")
   v
执行业务逻辑

1. LineBasedFrameDecoder

new LineBasedFrameDecoder(1024)

作用：

以 \n 或 \r\n 作为消息边界。
解决 TCP 字节流没有消息边界的问题。
收到完整一行后才向后传播。
默认删除行尾换行符。
限制单帧最大长度为 1024 字节。

例如：

接收的数据：

Hello\nWorld\n

拆分结果：

ByteBuf("Hello")
ByteBuf("World")

2. StringDecoder

new StringDecoder(StandardCharsets.UTF_8)

作用：

ByteBuf("Hello")
       |
       v
String("Hello")

StandardCharsets.UTF_8 属于 Java 8 自带的类。

StringDecoder 属于 Netty，需要引入 Netty 依赖。

3. StringEncoder

它是出站编码器，因此不会参与客户端消息的入站解码过程。

new StringEncoder(StandardCharsets.UTF_8)

作用：

String("OK\n")
       |
       v
ByteBuf([79, 75, 10])

4. MyServerHandler

public class MyServerHandler
        extends SimpleChannelInboundHandler<String> {

    @Override
    protected void channelRead0(
            ChannelHandlerContext ctx,
            String msg) {

        System.out.println("收到客户端消息：" + msg);

        ctx.writeAndFlush("OK\n");
    }
}

收到 Hello 后，执行业务逻辑，并向客户端回复 OK\n。

注意：LineBasedFrameDecoder 不会自动为出站数据添加换行符。

四、ctx.fireChannelRead() 的作用

ctx.fireChannelRead(msg);

作用：将入站消息继续传递给后面的 InboundHandler。

例如：

Head
 |
 v
Inbound A
 |
 | ctx.fireChannelRead(msg)
 v
Inbound B
 |
 | ctx.fireChannelRead(msg)
 v
Inbound C
 |
 v
Tail

代码示例：

public class HandlerA extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        System.out.println("Handler A 收到消息");

        // 继续传播入站事件
        ctx.fireChannelRead(msg);
    }
}

如果不调用 ctx.fireChannelRead(msg)，这个入站消息就不会继续向后传播。

注意：

ChannelInboundHandlerAdapter 不会自动释放 ByteBuf。
如果消息已经被消费且不继续传播，需要正确释放。
SimpleChannelInboundHandler 默认会自动释放它匹配并处理的引用计数消息。

五、ctx.writeAndFlush() 和 channel.writeAndFlush() 的区别

假设 Pipeline：

Head
 |
 A：StringEncoder       Outbound
 |
 B：MyServerHandler     Inbound
 |
 C：OtherEncoder        Outbound
 |
Tail

1. ctx.writeAndFlush()

在 B 中执行：

ctx.writeAndFlush("Hello");

执行路径：

B
|
| 向 Head 方向寻找 OutboundHandler
v
A：StringEncoder
|
v
Head
|
v
Socket 写出

因此：

A 会执行。
C 不会执行。
B 的入站方法不会因为这次写入而重新执行。

原因：ctx.writeAndFlush() 从当前 Context 向前传播。

2. channel.writeAndFlush()

在 B 中执行：

ctx.channel().writeAndFlush("Hello");

执行路径：

Tail
 |
 v
C：OtherEncoder
 |
 v
B：MyServerHandler（跳过）
 |
 v
A：StringEncoder
 |
 v
Head

因此：

C 会执行。
A 也会执行。
B 的入站处理方法被跳过。

注意：这是 Handler 执行路径示意，并不保证两个编码器的类型一定兼容。

3. 两种方法对比

方法	传播起点	方向
ctx.fireChannelRead(msg)	当前 Context	向 Tail
ctx.writeAndFlush(msg)	当前 Context	向 Head
ctx.channel().writeAndFlush(msg)	Tail	向 Head
ctx.pipeline().writeAndFlush(msg)	Tail	向 Head

推荐方式：

通常在业务 Handler 中使用 ctx.writeAndFlush()，并确保所需的 OutboundHandler 注册在当前 Handler 前面。

如果需要从 Pipeline 尾部经过完整的出站处理链，可以使用 channel.writeAndFlush()。

六、Handler 执行顺序

假设 Pipeline：

ch.pipeline().addLast(new HandlerA()); // Inbound
ch.pipeline().addLast(new HandlerB()); // Outbound
ch.pipeline().addLast(new HandlerC()); // Inbound
ch.pipeline().addLast(new HandlerD()); // Outbound

客户端发送一条消息。

HandlerA 执行：

ctx.fireChannelRead(msg);

HandlerC 收到消息后执行：

ctx.writeAndFlush(msg);

完整过程：

Head
 |
 v
HandlerA（Inbound）
 |
 | fireChannelRead(msg)
 v
HandlerB（Outbound）跳过
 |
 v
HandlerC（Inbound）
 |
 | writeAndFlush(msg)
 |
 | 开始向 Head 传播
 v
HandlerB（Outbound）
 |
 | ctx.write(msg, promise)
 v
Head
 |
 v
Socket

最终执行顺序：

HandlerA
    ↓
HandlerC
    ↓
HandlerB
    ↓
Socket 写出

HandlerD 不会执行，因为它位于 HandlerC 后面，而 ctx.writeAndFlush() 只向前查找 OutboundHandler。

注意：若实际发送的是 String，还必须有合适的字符串编码器，否则不能直接作为 Socket 字节数据写出。

七、Pipeline 底层实现原理

Netty Pipeline 内部通过 ChannelHandlerContext 组织成双向链表。

简化结构：

HeadContext
     ⇅
Context A
     ⇅
Context B
     ⇅
Context C
     ⇅
TailContext

每个 Context 保存：

当前的 ChannelHandler。
前一个节点的引用 prev。
后一个节点的引用 next。
对应的事件执行器信息。

例如：

ctx.fireChannelRead(msg);

Netty 沿 next 方向寻找能够处理入站事件的 Context。

ctx.writeAndFlush(msg);

Netty 沿 prev 方向寻找能够处理出站写入事件的 Context。

它不是简单调用所有 Handler，而是根据 Handler 的类型及事件传播方向选择执行节点。

另外，Pipeline 事件可能需要切换到目标 Handler 绑定的 EventExecutor 中执行，因此传播不一定始终发生在同一个调用线程中。

八、常见错误与注意事项

错误一：认为消息进入 Pipeline 后会执行所有 Handler。

正确理解：需要区分入站和出站处理器。

错误二：认为 StringDecoder 后面一定执行 StringEncoder。

正确理解：StringDecoder 处理入站字节，StringEncoder 处理出站数据。

错误三：认为 ctx.writeAndFlush() 会从 Pipeline 最后面开始传播。

正确理解：它从当前 Context 开始向 Head 寻找出站处理器。

错误四：认为 channel.writeAndFlush() 和 ctx.writeAndFlush() 完全相同。

正确理解：两者传播起点不同，经过的出站 Handler 可能不同。

错误五：重写 channelRead() 后不传递消息，也不释放 ByteBuf。

正确理解：如果要继续传播，调用 ctx.fireChannelRead(msg)；如果消费消息，需要正确管理引用计数。

错误六：使用 LineBasedFrameDecoder 后，认为出站消息会自动加换行符。

正确理解：拆帧器仅处理入站消息，出站消息需要自己添加换行符或使用对应编码器。

九、重点总结

1. ChannelPipeline 是什么？

ChannelPipeline 是 Netty 的 Handler 处理链，每个 Channel 都有自己的 Pipeline，用于处理入站和出站事件。

2. Inbound 和 Outbound 的区别？

Inbound 处理接收数据和连接生命周期事件，通常从 Head 向 Tail 传播；Outbound 处理发送、关闭等操作，向 Head 传播。

3. ctx.fireChannelRead() 的作用？

将当前入站消息继续传播给后面的 InboundHandler。

4. ctx.writeAndFlush() 与 channel.writeAndFlush() 的区别？

前者从当前 Context 向 Head 传播，后者从 Pipeline 的 Tail 向 Head 传播。

5. Pipeline 底层是什么结构？

由 ChannelHandlerContext 组成的双向链表，通过 prev 和 next 引用查找符合事件类型的 Handler。

6. 为什么 StringEncoder 通常放在业务 Handler 前面？

因为业务 Handler 使用 ctx.writeAndFlush() 时，出站事件只会向前传播，编码器放在前面才能被执行。

十、核心总结

Netty Pipeline 是双向 Handler 处理链：Inbound 向后传播，Outbound 向前传播；ctx 从当前位置出发，channel 从 Pipeline 尾部发起出站操作。

理解事件传播方向、Handler 类型以及 Context 的传播起点，是正确使用 Netty 编解码和业务处理器的基础。

Cloud_Drift's Blog

ChannelPipeline 与 Handler 事件传播机制

一、ChannelPipeline 是什么？

二、Inbound 和 Outbound 的区别

三、消息入站处理流程

1. LineBasedFrameDecoder

2. StringDecoder

3. StringEncoder

4. MyServerHandler

四、ctx.fireChannelRead() 的作用

五、ctx.writeAndFlush() 和 channel.writeAndFlush() 的区别

1. ctx.writeAndFlush()

2. channel.writeAndFlush()

3. 两种方法对比

六、Handler 执行顺序

七、Pipeline 底层实现原理

八、常见错误与注意事项

九、重点总结

十、核心总结

评论

Cloud_Drift's Blog

ChannelPipeline 与 Handler 事件传播机制

一、ChannelPipeline 是什么？

二、Inbound 和 Outbound 的区别

三、消息入站处理流程

1. LineBasedFrameDecoder

2. StringDecoder

3. StringEncoder

4. MyServerHandler

四、ctx.fireChannelRead() 的作用

五、ctx.writeAndFlush() 和 channel.writeAndFlush() 的区别

1. ctx.writeAndFlush()

2. channel.writeAndFlush()

3. 两种方法对比

六、Handler 执行顺序

七、Pipeline 底层实现原理

八、常见错误与注意事项

九、重点总结

十、核心总结

支持与分享

评论