ChannelPipeline 与 Handler

一句话：Pipeline 是双向链表，入站从 Head 走到 Tail，出站从 Tail 走到 Head；ctx.write 从当前节点向 Head 找 Outbound，channel.write 从 Tail 走完整条链。

适用版本：Netty 4.1.x。

为什么需要

网络事件有两类方向：读进来的数据要经过解码、鉴权、业务处理；写出去的数据要经过编码、加密、统计。如果每个 Handler 自己判断「我该处理什么、下一个是谁」，代码会迅速失控。

Pipeline 用一条链把「顺序」和「方向」固定下来：你只写自己关心的 Handler，Netty 负责调用下一个。

核心机制

数据结构：双向链表

Pipeline 是一条双向链表，入站事件从 Head 走到 Tail，出站从 Tail 走回 Head：

每个 ChannelHandlerContext 持有 Handler、前驱/后继指针、所属 Channel/Pipeline。双向链表让两个方向都能 O(1) 前进；数组只适合单向遍历。

入站与出站

入站（Inbound）：Head → Tail。读事件、解码、业务处理。
出站（Outbound）：Tail → Head。编码、加密、写 Socket。

Handler 只实现自己关心的方向：

ChannelInboundHandler：channelRead、channelReadComplete、userEventTriggered 等。
ChannelOutboundHandler：write、flush、connect、close 等。
ChannelDuplexHandler：两者都实现。

不关心的方向会被自动跳过，不需要空实现。

`ctx.write` 与 `channel.write` 的区别

调用	起点	经过的 Outbound
`ctx.write(msg)`	当前 Handler 的 Context	从当前节点向 Head 方向，经过它之前的 Outbound Handler
`ctx.channel().write(msg)`	Tail	从 Tail 向 Head，经过整条链上的所有 Outbound Handler

「之前」指 Pipeline 中靠近 Head 的一侧。以这条链为例（O 出站、I 入站）：

在 Business 里 ctx.write(msg)：只经过 Encoder，不经过 AccessLog。
在 Business 里 ctx.channel().write(msg)：从 Tail 开始，先经过 AccessLog，再经过 Encoder。

所以两者的区别不是「效率」，而是「要不要经过当前节点之后的 Outbound Handler」。需要完整管线处理（比如统一加日志/加密）用 channel.write；只走剩余编码用 ctx.write。

Head 与 Tail

Head：连接 Pipeline 与 Socket。入站时从 Socket 读数据并 fireChannelRead；出站时最终调用 unsafe.write() 写 Socket。
Tail：入站的最后一站。未被释放的消息会在这里被释放并打警告，防止内存泄漏。

异常与传播

ctx.fireExceptionCaught(cause) 把异常传给下一个 Inbound Handler。
如果一直没人处理，Tail 会打警告；连接不会自动关闭，是否关闭由业务决定。
出站异常通常以 Promise 失败的形式返回，不经过 exceptionCaught。

实战场景

一、请求进得来、响应回不去：入站链被中途截断

现象：IM 网关灰度后，约 0.3% 的消息客户端发出去就没有任何响应（经验值，需校准）；服务端 access 日志里能看到这条连接，但业务处理没有痕迹。同时 Error-prone/ResourceLeakDetector 日志出现 LEAK: ByteBuf.release() was not called。

排查：入站事件不会自己跳到下一个节点，谁接了就得决定往下传不往下传。

定位层	取证据	判定
链上顺序	连接建立时打印 `ctx.pipeline().names()`	与预期装配顺序不符 → 装配阶段就错
事件断在哪	在每个 Inbound Handler 入口打 `msg` 类型与 hash	某节点之后不再出现 → 该节点吞了事件
Handler 基类	是否用 `ChannelInboundHandlerAdapter` 且没调 `ctx.fireChannelRead(msg)`	Adapter 不自动传递，也不自动释放
分支覆盖	`if/else` 是否有未 `fire` 的分支（未知类型、鉴权失败）	只在特定消息类型丢 → 分支问题
异常路径	`exceptionCaught` 里是否 `ctx.close()`	未关闭时连接存在但永不响应，最难发现

本例命中的是新增的限流 Handler：超阈值时只做了 return，既没有 fireChannelRead，也没有释放 ByteBuf。于是消息静默消失，泄漏检测同时报警。

对症做法：

java

// 1. 规则：入站 Handler 要么传递，要么消费并释放，不允许第三种出口
@Override public void channelRead(ChannelHandlerContext ctx, Object msg) {
    if (overLimit(ctx)) {
        ReferenceCountUtil.release(msg);                       // 消费即负责释放
        ctx.writeAndFlush(Response.reject("RATE_LIMITED"));    // 给客户端确定性失败
        return;
    }
    ctx.fireChannelRead(msg);                                  // 否则必须显式传递
}

// 2. 优先用 SimpleChannelInboundHandler：返回后自动释放，且类型不匹配会自动往下传
public class BizHandler extends SimpleChannelInboundHandler<Frame> {
    @Override protected void channelRead0(ChannelHandlerContext ctx, Frame f) { /* 不用释放 */ }
}

// 3. 异常必须留下痕迹，而不是静默持有连接
@Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
    log.warn("channel fail, remote={}", ctx.channel().remoteAddress(), cause);
    ctx.close();
}

验证方式很简单：在 Tail 之前挂一个统计 Handler，记录“进入业务链的消息数”与“离开业务链的消息数”；两者不等就说明中间有人吞了事件或消息。

复盘要点：把“入站 Handler 的出口只有两种”写进代码模板，并用一条静态检查约束（channelRead 方法体里必须出现 fireChannelRead 或 release）。这类问题的代价不是慢，而是客户端完全无法感知——静默失败比错误更难查。

二、`ctx.write` 没经过 Encoder：写出类型不对

现象：压测某个特定接口时报 writeObject(...) failed. message type: java.lang.String, expected: ByteBuf，其他接口正常；错误只在业务 Handler 内部主动写响应的路径上出现。

排查：ctx.write 从当前节点向 Head 方向找 Outbound，不经过它之后（靠近 Tail 一侧）的 Encoder；channel.write 从 Tail 开始走完整条链。

定位层	取证据	判定
Encoder 位置	打印 `ctx.pipeline().names()`，看 Encoder 在业务 Handler 的哪一侧	Encoder 靠近 Tail → `ctx.write` 到不了它
调用方式	出错代码用的是 `ctx.write` 还是 `channel.write`	`ctx` 起点为当前节点，`channel` 起点为 Tail
正常路径为何不报错	其他接口是否由别的 Handler 写出	写出点不同，经过的 Outbound 就不同
Head 的报错	栈里出现 `AbstractChannel$AbstractUnsafe.write` 的类型检查	说明没有任何 Encoder 把它转成 ByteBuf

出站方向是“从 Tail 往 Head”，按添加顺序去想必然出错：Encoder 写在业务之后，业务用 ctx.write 就绕过了它，String 直接被送到 Head 做 socket 写，类型检查失败。

对症做法：

java

// 1. 装配期固定顺序：FrameDecoder → 解码 → 业务 → Encoder（Encoder 靠近 Head 一侧）
ch.pipeline()
  .addLast(new LengthFieldBasedFrameDecoder(...))
  .addLast(new RequestDecoder())
  .addLast("encoder", new ResponseEncoder())      // 放在业务之前，保证 ctx.write 能经过
  .addLast(new BizHandler());

// 2. 想让写出经过完整出站链（统一加密、埋点、审计），显式从 channel 入口写
ctx.channel().writeAndFlush(response);

// 3. 装配顺序做成启动断言，而不是靠 review
List<String> names = pipeline.names();
assertThat(names.indexOf("encoder")).isLessThan(names.indexOf("biz"));

选择依据不是“哪个更快”，而是“要不要经过当前节点之后的 Outbound Handler”：需要完整出站处理用 channel.write，只走剩余编码用 ctx.write。

复盘要点：Pipeline 顺序错误属于结构性错误，应当在启动时或测试里断住。给每个协议装配一个 PipelineAssembler，把顺序作为唯一入口并暴露 names 列表供自检；同类问题还会出现在 TLS（SslHandler 必须在最前）和压缩 Handler 上，见拆包粘包与编解码。

三、`@Sharable` 有状态 Handler：连接之间串话

现象：文件传输服务在并发传两个文件时，客户端 A 收到 B 的分片，校验失败；单连接串行传输一切正常。代码里 TransferHandler 是 new 出来后 addLast 的，且带了 @Sharable。

排查：一个 Handler 实例被多个 Channel 共用，状态就会互相覆盖。

定位层	取证据	判定
实例是否共享	打印 `System.identityHashCode(handler)` 与各 Channel 的对应关系	多连接同一 hash → 共享实例
是否有状态	类里是否有 `StringBuilder`、`currentFile`、`offset` 等实例字段	有状态 + 共享 = 必然串
注解语义	是否标了 `@Sharable`	该注解只是声明“可安全共享”，不会帮你隔离
线程因素	同一 EventLoop 串行也不会隔离连接	串话来自实例共享，不是并发写

对症做法：

java

// 1. 有状态 Handler 每连接新建（放在 initChannel 里，而不是外面 new）
.childHandler(new ChannelInitializer<SocketChannel>() {
    protected void initChannel(SocketChannel ch) {
        ch.pipeline().addLast(new TransferHandler(sessionStore));   // 每连接一个实例
    }
});

// 2. 无状态 Handler 才标 @Sharable，字段只允许注入的协作对象
@Sharable public class MetricsHandler extends ChannelInboundHandlerAdapter { /* 无实例状态 */ }

// 3. 跨连接的状态放外部容器，按 Channel 归属，而不是塞进 Handler 字段
class Session { final Channel channel; long offset; String fileId; }
private final Map<ChannelId, Session> sessions = new ConcurrentHashMap<>();

// 4. 群发/广播用 ChannelGroup，它自己处理连接生命周期
ChannelGroup room = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
room.addAndFlush(msg);

java

// 5. 断言测试：两个并发连接的状态必须互不影响
@Test void concurrentTransfersDoNotShareState() {
    EmbeddedChannel a = new EmbeddedChannel(new TransferHandler(store));
    EmbeddedChannel b = new EmbeddedChannel(new TransferHandler(store));
    a.writeInbound(frame("fileA", 0));
    b.writeInbound(frame("fileB", 0));
    assertThat(readOut(a).fileId()).isEqualTo("fileA");
    assertThat(readOut(b).fileId()).isEqualTo("fileB");
}

EmbeddedChannel 是这类问题最便宜的验证手段：不需要真实网络，就能同时持有两个连接的独立 Pipeline。

复盘要点：@Sharable 的语义是“我保证无状态”，不是“Netty 帮我隔离”。判据可以机械化：带实例可变字段的 Handler 类禁止 @Sharable，用 ArchUnit 规则挡住；状态归属统一到 Session + ChannelGroup，让 Handler 只做事件搬运。

最小上手

用换行分帧做一个 echo 服务，验证 Pipeline 顺序：

java

public class PipelineServer {
    public static void main(String[] args) throws Exception {
        EventLoopGroup boss = new NioEventLoopGroup(1);
        EventLoopGroup worker = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(boss, worker)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) {
                     ch.pipeline()
                       .addLast(new LineBasedFrameDecoder(1024))
                       .addLast(new StringDecoder(StandardCharsets.UTF_8))
                       .addLast(new StringEncoder(StandardCharsets.UTF_8))
                       .addLast(new SimpleChannelInboundHandler<String>() {
                           @Override
                           protected void channelRead0(ChannelHandlerContext ctx, String msg) {
                               ctx.writeAndFlush("echo:" + msg + "\n");
                           }
                       });
                 }
             });
            b.bind(8080).sync().channel().closeFuture().sync();
        } finally {
            boss.shutdownGracefully();
            worker.shutdownGracefully();
        }
    }
}

验证：

bash

printf 'hi\n' | nc 127.0.0.1 8080
# 期望 echo:hi

这里 StringEncoder 在业务 Handler 之前（靠近 Head），所以 ctx.writeAndFlush 从业务节点向 Head 走时会经过它。如果把 Encoder 放到业务 Handler 之后（靠近 Tail），ctx.write 就不会经过它，写出去的是 String，最终在 Head 报类型错误。

生产动作

顺序检查：FrameDecoder 必须在最前；业务解码器在 FrameDecoder 之后；Encoder 要在业务 Handler 靠近 Head 的一侧（或明确用 channel.write）。
自动释放：SimpleChannelInboundHandler 会在 channelRead0 返回后释放消息；ChannelInboundHandlerAdapter 不会，需要自己 ReferenceCountUtil.release。
@Sharable：只有无状态 Handler 才能复用；有状态的 Handler 每个 Channel 新建实例，否则会出现串话。
不要阻塞：Pipeline 回调在 EventLoop 线程执行；耗时逻辑丢业务线程池，再 ctx.executor().execute(...) 写回。
动态增删：运行时 addLast/remove 要在这个 Channel 的 EventLoop 上执行；跨线程操作要 ctx.executor().execute(...)。
日志：在 Head 之后、业务之前加一个 Inbound 日志 Handler，能看到解码前/后的差异；不要在业务 Handler 里打全量 payload。

排障：

消息丢了 → 查是否忘了 ctx.fireChannelRead(msg) → 在 Tail 前加日志确认。
写出去类型不对 → 查 Encoder 是否在写路径上 → 对比 ctx.write 与 channel.write。
内存泄漏 → 查 Handler 是否把 ByteBuf 存下来却没 retain，或 Adapter 没 release。

边界

不是业务编排引擎：Pipeline 适合协议处理链，不适合复杂业务流程；业务流程放 Service 层。
Handler 有状态就不要共享：@Sharable 只适合无状态；有状态共享会导致并发问题。
出站顺序容易反直觉：记住「从 Tail 往 Head」，不要按添加顺序想。
不要在 Pipeline 里做重活：它是 IO 线程的一部分；重活会拖慢该 EventLoop 上的所有连接。
Tail 不是万能兜底：它只释放走到 Tail 的消息；被中途吞掉的消息不会到 Tail。

讲清楚

Pipeline 是一条双向链表，入站从 Head 到 Tail，出站从 Tail 到 Head。Handler 只实现自己关心的方向，不关心的自动跳过。ctx.write 从当前节点向 Head 找 Outbound Handler，channel.write 从 Tail 开始走完整条链；前者适合只走剩余编码，后者适合需要完整出站处理。Head 负责和 Socket 交互，Tail 负责兜底释放。最常见的 bug 是入站 Handler 忘了 fireChannelRead，消息被截断；以及 Encoder 放错位置，ctx.write 没经过它，写出去的类型不对。

Spring 微服务

Netty

MyBatis

设计模式 · 导读

Spring 框架

DDD 领域驱动设计 · 导读

Nacos

ChannelPipeline 与 Handler

为什么需要

核心机制

数据结构：双向链表

入站与出站

`ctx.write` 与 `channel.write` 的区别

Head 与 Tail

异常与传播

实战场景

一、请求进得来、响应回不去：入站链被中途截断

二、`ctx.write` 没经过 Encoder：写出类型不对

三、`@Sharable` 有状态 Handler：连接之间串话

最小上手

生产动作

边界

讲清楚

ChannelPipeline 与 Handler ​

为什么需要 ​

核心机制 ​

数据结构：双向链表 ​

入站与出站 ​

ctx.write 与 channel.write 的区别 ​

Head 与 Tail ​

异常与传播 ​

实战场景 ​

一、请求进得来、响应回不去：入站链被中途截断 ​

二、ctx.write 没经过 Encoder：写出类型不对 ​

三、@Sharable 有状态 Handler：连接之间串话 ​

最小上手 ​

生产动作 ​

边界 ​

讲清楚 ​

ChannelPipeline 与 Handler

为什么需要

核心机制

数据结构：双向链表

入站与出站

`ctx.write` 与 `channel.write` 的区别

Head 与 Tail

异常与传播

实战场景

一、请求进得来、响应回不去：入站链被中途截断

二、`ctx.write` 没经过 Encoder：写出类型不对

三、`@Sharable` 有状态 Handler：连接之间串话

最小上手

生产动作

边界

讲清楚