【Netty源码解读和权威指南】第32篇：Netty背压机制——不让发送方“撑死“接收方

上一篇【第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

publicclassBackPressureHandlerextendsChannelInboundHandlerAdapter{privatevolatilebooleanpaused;privateQueue<Object>buffer=newConcurrentLinkedQueue<>();@OverridepublicvoidchannelRead(ChannelHandlerContextctx,Objectmsg){if(paused){buffer.offer(msg);// 暂存}else{ctx.fireChannelRead(msg);// 正常传播}// 发送响应并检查水位线ChannelFuturef=ctx.writeAndFlush(process(msg));if(!ctx.channel().isWritable()){f.addListener(future->{paused=true;// 暂停从上游拉取数据upstreamSource.pause();});}}@OverridepublicvoidchannelWritabilityChanged(ChannelHandlerContextctx){if(ctx.channel().isWritable()){paused=false;upstreamSource.resume();// 处理积压消息Objectmsg;while((msg=buffer.poll())!=null){ctx.fireChannelRead(msg);}}ctx.fireChannelWritabilityChanged();}}

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

publicclassBackPressureProxyextendsChannelInboundHandlerAdapter{privateChannelbackendChannel;privatebooleanfrontendPaused;@OverridepublicvoidchannelRead(ChannelHandlerContextctx,Objectmsg){if(backendChannel!=null&&backendChannel.isActive()){backendChannel.writeAndFlush(msg).addListener(f->{// 检查后端是否满if(!backendChannel.isWritable()&&!frontendPaused){frontendPaused=true;ctx.channel().config().setAutoRead(false);// 暂停读前端}});}}@OverridepublicvoidchannelWritabilityChanged(ChannelHandlerContextctx){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连接管理与心跳检测——工业级断连处理方案