上一篇【第31篇】Netty零拷贝深度解析——性能极致的秘密武器
下一篇【第33篇】Netty连接管理与心跳检测——工业级断连处理方案

开篇故事

某支付系统，生产者TQ是消费者的10倍，结果消费者OOM了——消息发送太快，来不及处理！

背压（Back Pressure）：下游处理不过来时，通知上游减速。

Netty天然支持背压：通过ChannelOutboundBuffer的水位线+TCP滑动窗口。

一、背压的TCP层基础

TCP滑动窗口：
发送方                          接收方
  |                               |
  |--数据1~1000------------------>|
  |--数据1001~2000--------------->|
  |                         接收窗口满！
  |<-------ACK(win=0)------------|  ← 窗口为0，暂停发送
  |                         处理完成
  |<-------ACK(win=5000)---------|  ← 窗口恢复，继续发送
  |--数据2001~3000--------------->|

二、Netty应用层背压

2.1 水位线机制

// 默认配置
writeBufferHighWaterMark = 64KB   // 高水位
writeBufferLowWaterMark = 32KB    // 低水位

// 状态转换
if (pendingBytes > highWaterMark && writable) {
    writable = false;   // 不可写
    fireChannelWritabilityChanged();  // 触发事件
}
if (pendingBytes < lowWaterMark && !writable) {
    writable = true;    // 可写
    fireChannelWritabilityChanged();  // 触发事件
}

2.2 背压处理Handler

public class BackPressureHandler extends ChannelInboundHandlerAdapter {
    private volatile boolean paused;
    private Queue<Object> buffer = new ConcurrentLinkedQueue<>();
    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (paused) {
            buffer.offer(msg);  // 暂存
        } else {
            ctx.fireChannelRead(msg);  // 正常传播
        }
        // 发送响应并检查水位线
        ChannelFuture f = ctx.writeAndFlush(process(msg));
        if (!ctx.channel().isWritable()) {
            f.addListener(future -> {
                paused = true;
                // 暂停从上游拉取数据
                upstreamSource.pause();
            });
        }
    }
    
    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) {
        if (ctx.channel().isWritable()) {
            paused = false;
            upstreamSource.resume();
            // 处理积压消息
            Object msg;
            while ((msg = buffer.poll()) != null) {
                ctx.fireChannelRead(msg);
            }
        }
        ctx.fireChannelWritabilityChanged();
    }
}

三、实战：带背压的代理服务器

public class BackPressureProxy extends ChannelInboundHandlerAdapter {
    private Channel backendChannel;
    private boolean frontendPaused;
    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (backendChannel != null && backendChannel.isActive()) {
            backendChannel.writeAndFlush(msg).addListener(f -> {
                // 检查后端是否满
                if (!backendChannel.isWritable() && !frontendPaused) {
                    frontendPaused = true;
                    ctx.channel().config().setAutoRead(false);  // 暂停读前端
                }
            });
        }
    }
    
    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) {
        if (ctx.channel().isWritable() && frontendPaused) {
            frontendPaused = false;
            ctx.channel().config().setAutoRead(true);  // 恢复读前端
        }
        ctx.fireChannelWritabilityChanged();
    }
}

四、背压链路图

客户端→Netty前端Handler→Netty后端Handler→下游服务
         ↑                                    ↑
    autoRead=false                    channelWritabilityChanged
    (暂停读取)                           (恢复读取)
         |                                    |
         +--------→ 水位线触发 ←-------------+

五、总结

上一篇【第31篇】Netty零拷贝深度解析——性能极致的秘密武器
下一篇【第33篇】Netty连接管理与心跳检测——工业级断连处理方案