Netty源码系列写到第十一篇终于轮到写数据。前面聊读数据、pipeline、解码器的时候总有人问那写出去的数据Netty到底是怎么处理的其实不管你是做RPC框架返回响应、网关转发请求还是推送服务往下游写消息最终都逃不过write和flush这两个动作。很多人对Netty的印象就是异步、高性能但真正去看源码时往往卡在出站这条链路上——它跟入站的处理方式完全不一样涉及缓冲区、水位线、背压、NIO底层写入等一堆概念。这篇文章就把这条链路从入口到内核层完全拆开把写数据的核心源码逐段讲透顺便把我在实际项目中踩过的坑一起倒出来。1. 写数据的前置链路从一行代码开始1.1 两个入口write与flush的职责划分Netty的写操作有两个入口使用方式完全不同这是很多初学者第一个踩坑的地方。// 方式A从Channel入口写 channel.write(msg); // 方式B从ChannelHandlerContext入口写 ctx.write(msg);两者的核心区别在于事件传播的起点不同。channel.write()的调用会从tailContext开始往前找也就是pipeline中最后一个ChannelOutboundHandler往前传播而ctx.write()是从当前handler位置往前找outbound处理器。这直接决定了你的消息能经过哪些编码器、哪些业务处理器。我见过不少人在ChannelInboundHandler里直接写channel.write()结果消息绕过了pipeline尾部后续的outbound处理器导致该做的编码没做或者统计逻辑没生效。这也是Netty官方一直强调的如果你在某个ChannelHandler内部需要写出站数据最好用ctx.write()而不是channel.write()这样能保证消息从当前节点的下一个outbound节点继续传播。再说write和flush的分工。write只是把消息包装成任务入队存进ChannelOutboundBuffer并不会真正触发网络写入。只有flush才会把缓冲区的数据刷给操作系统。为什么要做这种分离为了批量合并把多次write的数据攒在一起一次flush刷出去减少系统调用次数。这是Netty高性能的一个关键点你想想如果业务线程每写一条消息就触发一次write系统调用那和BIO的write有啥区别。所以正确的使用姿势是多次write最后统一flush。Netty的writeAndFlush只是这个过程的语法糖底层还是先走write再走flush。1.2 从入口到缓冲区的传递路径我们来看write事件在pipeline中是怎么传播的。以ctx.write(msg)为例AbstractChannelHandlerContext.write方法是整个出站事件的起点private void write(Object msg, boolean flush, ChannelPromise promise) { AbstractChannelHandlerContext next findContextOutbound(flush ? MASK_WRITE_AND_FLUSH : MASK_WRITE); ... next.invokeWrite(msg, promise); ... }findContextOutbound会从当前节点往前找找到第一个outbound类型的节点。找到之后调用invokeWrite一路向前传播直到headContext。整个过程就是一个反向遍历pipeline的过程跟入站事件从head往tail传播正好相反。最终在headContext中write被转换成unsafe.write进而调用ChannelOutboundBuffer.addMessage把消息正式加入出站缓冲区。到这里数据暂时入队还没真正发送真正的网络写入发生在flush到headContext之后。这条链路咱们先记住后面会一步步拆。2. 核心数据结构ChannelOutboundBuffer如何管住出站数据2.1 Entry链表与三指针滑窗写数据这条链路最核心的数据结构就是ChannelOutboundBuffer它的内部维护了一个由Entry节点组成的链表加上三个指针flushedEntry第一个待刷出的节点、unflushedEntry第一个未刷出的节点、tailEntry链表尾节点。为了方便理解你可以把ChannelOutboundBuffer想象成一条传送带write是往传送带上放货flush是把传送带前端已经放好的货推出去。已入队但还没调用flush的消息属于未刷出区间一旦调用了flush这些消息就进入待刷出区间flushedEntry会指向其中的第一个节点。每个Entry内部保存的是消息本身、对应的ChannelPromise、以及预估内存大小pendingSize。注意这个pendingSize不是消息的真实字节数而是Netty估算出来的内存占用后面水位线计算依赖的就是这个值。Entry本身还被Netty做了对象复用设计从RECYCLER中获取用完了归还到对象池这也是为什么你在写代码时总是看到不要持有Entry引用这样的忠告——对象被回收复用后你持有的引用可能指向一个完全不同的数据。我建议阅读这块源码时重点看三个方法addMessage入队、remove刷出成功后移除、removeBytes部分字节刷出后更新进度。这三个方法把整个缓冲区的生命周期串了起来。2.2 估算内存与水位线ChannelOutboundBuffer里有几个关键字段其中totalPendingSize是累计的预估内存大小每次addMessage和remove都会更新它。还有一个unwritable字段用于标记当前缓冲区是否可写。这个可写不可写的判定逻辑非常重要直接关系到背压机制能否生效。private void incrementPendingOutboundBytes(long size) { if (size 0) { return; } long newWriteBufferSize TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, size); if (newWriteBufferSize channel.config().getWriteBufferWaterMark().high()) { setUnwritable(); } }这里的size是消息在缓冲区中所占的估算值具体规则是直接内存的ByteBuf按实际容量计算堆内存的ByteBuf按容量的2倍计算因为要考虑GC复制开销。水位线默认高水位是64KB低水位是32KB。当totalPendingSize超过高水位setUnwritable会把unwritable置为true并触发fireChannelWritabilityChanged事件通知pipeline上的handler当前通道暂时不可写。这个机制我特别想多说一句因为很多人的背压失效问题就出在这里。incrementPendingOutboundBytes的size估算不是精确的不同内存类型、不同消息大小算出来的值和真实字节数是有出入的。如果你在业务里依赖这个值做精确控制一定会出问题。我用过的一个项目就出现过这种情况每个消息的预估size远小于实际size导致高水位一直不触发但实际内存已经被消息堆满最后直接OOM。排查下来发现是业务side用了大量复合缓冲区而复合缓冲区的估算规则和普通缓冲区不一样在addMessage时算的是内部每个组件的总和看似没问题但这个总和和最终写入网络的实际字节数可能有较大偏差。后面会专门讲这个坑。3. write()全链路源码拆解3.1 入队逻辑addMessage到底做了什么前面我们说到headContext.write最终会调用unsafe.write真正干活的是AbstractUnsafe.write它会代理到ChannelOutboundBuffer.addMessagepublic void write(Object msg, ChannelPromise promise) { ... outboundBuffer.addMessage(msg, safeConvertPendingSize(...), promise); ... }addMessage做的事情很纯粹新建或复用Entry把消息和promise塞进去挂到链表尾部更新tailEntry再累加totalPendingSize。还有一个关键细节它调用了safeConvertPendingSize做了一次大小转换防止溢出或者出现负数。我只讲一个容易忽略的点write操作如果发生在EventLoop之外的业务线程会先被封装成一个WriteTask提交到EventLoop的任务队列由IO线程执行addMessage。这就解释了为什么channel.write()是线程安全的即使你在业务线程池里调用最终消息入队和后续刷出都是由单线程的EventLoop完成的天然避免并发竞争。但也正是这个特性导致Netty里有一个经典性能问题业务线程高频调用writeAndFlush每个操作都向EventLoop提交一个任务EventLoop忙于处理任务队列网络IO的响应延迟反而被拖高。这就是为什么官方文档反复说不要在IO线程中做耗时操作的同时也说不要把高频业务写入直接丢给EventLoop。3.2 消息的搭桥与合并addMessage入队时支持一个非常有用的特性把多个ByteBuf合并成一个复合缓冲区。Netty内部有一个CompositeByteBuf当你在代码里通过ctx.write(new DefaultCompositeByteBuf(...))写入时addMessage并不会拆开它而是先在缓冲区里做好相关标记等到flush阶段nioBuffers方法才会把这个复合缓冲区拆成一组NIO的ByteBuffer供底层socket.write(ByteBuffer[])批量写入。这个设计是为了配合NIO的Gathering Write操作它允许一次系统调用写入多个缓冲区本质上是writev。如果底层不支持Gathering WriteNetty会退化成逐个写。真正重要的点在于如果你要写出站数据尽量把多次write的ByteBuf通过CompositeByteBuf聚合成一次write而不是分别调用多次write这样既减少了Entry数量也减少了后续flush时系统调用的次数。不过这里也有个代价CompositeByteBuf的引用计费管理比普通ByteBuf更复杂。你组合了多个子缓冲区释放时如果不注意要么重复释放要么子缓冲区内存泄漏。我建议在业务逻辑中少手动构造CompositeByteBufNetty在flush阶段对多个连续已入队的ByteBuf本身就有合并写优化你只需要在一次flush前连续调用多次write。4. flush()全链路数据如何真正抵达操作系统4.1 flush事件的传播机制flush操作的传播路径跟write类似从调用方开始往前找到outbound处理器一路传播到headContext。在headContext中调用unsafe.flush最终来到AbstractUnsafe.flush0。protected void flush0() { ... in.invokeFlush0(); ... doWrite(in); ... }flush0里有个细节它会先调用doBeginRead对NIO来说就是重新注册OP_READ事件然后才调用doWrite。这么做是有原因的如果注册读事件时发现socket不可读Netty会先处理一次写优先避免在写操作前丢失读事件。这个顺序问题我曾经在一次高并发网关调优中遇到过当时的现象是某段时间内读事件处理明显变慢后来定位发现就是在flush0之前没有正确调用doBeginRead导致读事件一直处于挂起状态调整之后恢复了正常。接下来doWrite就是真正的底层写入逻辑不同类型的Channel有不同实现。我们重点看NioSocketChannel的doWrite。4.2 NIO doWrite的核心循环NioSocketChannel.doWrite是整个写数据链路中最核心的一段我建议你直接打开源码对比着看。简化后的大致逻辑是protected void doWrite(ChannelOutboundBuffer in) throws Exception { SocketChannel ch javaChannel(); int writeSpinCount config().getWriteSpinCount(); while (in.isWritable()) { if (in.isEmpty()) { clearOpWrite(); return; } int bufferCount in.nioBufferCount(); long written 0; boolean done false; boolean setOpWrite false; if (bufferCount 1) { // 单缓冲区写入 ByteBuffer buffer in.nioBuffers()[0]; int size in.size(); written ch.write(buffer); if (written 0) { in.removeBytes(written); done size 1; } } else { // 多缓冲区使用Gathering Write ByteBuffer[] buffers in.nioBuffers(); int size in.size(); written ch.write(buffers); if (written 0) { in.removeBytes(written); done size 1; } } if (written 0) { if (done) { in.remove(); continue; } } else { // 写不进去了socket发送缓冲区满 setOpWrite true; break; } if (--writeSpinCount 0) { continue; } setOpWrite true; break; } if (setOpWrite) { setOpWrite(); } }这段代码信息量很大我拆成几个重点讲。第一writeSpinCount是写自旋次数默认16次。它的作用是限制一次flush事件中最多调用多少次底层write系统调用避免单次flush事件占用IO线程太久导致其他channel得不到处理。如果你写的数据量很大一次flush没写完Netty会通过setOpWrite()注册OP_WRITE事件等socket发送缓冲区变得可写之后由Selector触发下一次写事件继续写。所以写大数据量时一次writeAndFlush实际上可能横跨多个IO轮询周期。第二两种写入路径的判断逻辑bufferCount 1走单缓冲写入否则走Gathering Writewritev。关键点是如果底层NIO不支持Gathering WritenioBufferCount()会返回1即使你有多个缓冲区Netty也会退化成逐个写。第三removeBytes和remove的区别要搞清楚。removeBytes(written)只移除已经成功写入的字节remove()移除整个Entry。当donetrue时才执行remove()代表当前Entry里的数据全部写完。这里有个隐藏的bug隐患如果written已经是正数但还没写完你错误的调用remove()把Entry移除那剩余未写的数据就直接丢了。我在自己写简化版网络框架时踩过这个坑排查了半天才发现是remove时机不对。还有一个很多人忽略的地方当ch.write(buffer)返回0时表示socket的发送缓冲区已经满了写不进去了。此时如果继续死循环重试会把CPU打满且没有任何进展所以Netty直接setOpWrite并break把控制权交还给Selector。这也是背压设计在NIO层面的一次体现。4.3 零拷贝与FileRegion如果写入的数据来自文件Netty提供了FileRegion机制让底层直接调用transferTo将文件内容发送到socket全程不经过用户态拷贝。这个机制的实现藏在NioSocketChannel.doWrite的另一个分支中当in.current()返回的是FileRegion时会调用FileRegion.transferTo这个函数内部走的是NIO的FileChannel.transferTo。零拷贝听起来很美好但使用时有几个坑要注意。第一FileRegion写入时如果你的消息还在pipeline里后续的编码器可能根本处理不了FileRegion因为FileRegion不是一个ByteBuf它是独立的抽象。第二FileRegion的释放方式跟ByteBuf不同它通过自身的release()方法来管理引用计数如果你在pipeline中把FileRegion包装成某些自定义对象release之路会变得很曲折。我在一个文件传输服务里用过FileRegion当时是直接把DefaultFileRegion写入channel然后底层自动走transferTo。好处是CPU占用率明显下降坏处是日志、指标采集这类中间处理器都需要单独处理FileRegion类型比普通ByteBuf麻烦不少。如果你只是传小文件或者数据本身在内存里老老实实用ByteBuf就好零拷贝不是万能的。5. 背压机制与写失败兜底5.1 高水位线下的可写状态切换前面已经提到了水位线这里展开讲一下完整机制。ChannelOutboundBuffer中的totalPendingSize累计了所有未刷出消息的预估大小当它超过writeBufferWaterMark.high()默认64KB时会调用setUnwritable()将unwritable置为true并触发fireChannelWritabilityChanged()。反过来当totalPendingSize降到low()默认32KB以下时又调用setWritable()恢复可写状态。这个机制的本质是让应用感知到底层socket的消化能力。当网络变慢、对端不读数据时socket发送缓冲区会逐渐占满Netty的写操作会被阻塞在OP_WRITE等待环节totalPendingSize就会一直积压一旦超过高水位你监听channelWritabilityChanged事件就能知道该停手了别继续发数据了。// 在业务handler中感知可写状态变化 Override public void channelWritabilityChanged(ChannelHandlerContext ctx, ChannelWritabilityChangedEvent evt) { if (!ctx.channel().isWritable()) { // 通道不可写停止发送 } ctx.fireChannelWritabilityChanged(evt); }这个机制我见过几个常见的滥用方式。一种是业务在不可写时压根不理会继续疯狂往通道里write导致内存无上限增长。另一种是低频业务误以为只要调用了writeAndFlush数据就一定发出去了实际上在不可写期间调用write只是往内存缓冲区堆积如果你不检查返回结果消息会一直在内存里积压直到OOM。正确做法是维护一个待发送队列当通道不可写时把消息存到业务自己的有界队列里等可写恢复后再发送。还有一点要提醒isWritable()判断的是totalPendingSize是否低于水位线这个值本身只是一个估算值并不精确反映底层socket发送缓冲区的真实占用。所以这个值更适合用来做粗糙的流量控制不适合做精确的背压判断。5.2 写异常与失败兜底写数据过程中如果发生IOException比如连接被重置、对端关闭、写缓冲区异常Netty的处理路径跟读数据不同。在doWrite抛出异常后异常会向上传播到pipeline中触发exceptionCaught事件。同时flush0会调用invokeClose或者shutdownOutput之类的收尾操作。这个过程中有一个非常隐蔽的坑当你向一个已经关闭的连接写入消息时消息对应的ChannelPromise永远不会被成功通知而是收到ClosedChannelException。如果你没有给write操作设置ChannelFutureListener去监听失败回调那么这个异常可能会被Netty吞掉因为promise的失败通知如果没有监听者异常就静默丢失了你根本不知道消息没发出去。我见过不少线上故障最终定位到原因就是这个——业务方调用writeAndFlush之后完全不关心future连接出问题时消息静默丢失上游还在傻傻等待响应。正确的姿势是channel.writeAndFlush(msg).addListener((ChannelFutureListener) future - { if (!future.isSuccess()) { // 处理失败比如记录日志、重试、释放资源 } });而且这个监听器尽量放在EventLoop外的业务线程或者一个独立线程池中执行不要在IO线程做耗时操作。另外在写失败后ChannelOutboundBuffer会进入一个特殊状态failFlushed和close会把这些失败的Entry逐一出队并通知对应的ChannelPromise失败。如果你的pipeline里有ChannelOutboundHandlerAdapter的write方法被覆写切记不要在覆写时吞掉promise的完成状态否则上游永远等不到回调。6. 实践中的坑与优化建议6.1 常见问题速查表现象根本原因排查思路内存持续增长写入过快缓冲区积压检查totalPendingSize是否持续在高水位之上确认是否背压生效写入操作返回成功但消息丢失promise失败被静默忽略给writeFuture添加监听器检查失败ThrowableEventLoop线程CPU飙高高频率writeAndFlush导致任务队列拥塞在业务线程批量写入减少EventLoop任务提交频率大数据量写入变慢socket发送缓冲区满触发OP_WRITE频繁切换检查对端消费能力增加合并写调整水位线Netty内存泄漏告警ByteBuf未正确release检查写入缓冲区中每个Entry是否在remove时被释放确认是否存在未flush的消息6.2 写性能优化建议基于源码的研读和实际项目的调优经验我整理了三条针对写数据的高性价比优化方向未必全面但每一条都在实际工作中验证过。第一减少write次数用批量写替代频繁小写。NIO层的Gathering Write一次可以写多个缓冲区但这要求你在一次flush前把消息都攒到ChannelOutboundBuffer里。如果你的业务本身就是高吞吐推送场景可以尝试合并多个逻辑消息到一个ByteBuf或者干脆在业务层做消息队列批量取、批量写、批量flush。我们之前在某网关项目上把一个高频小消息推送接口从每条消息单独writeAndFlush改成每50条合并write一次flush在同等流量下CPU占用率下降了将近三分之一GC压力也明显缓解。第二动态调整水位线而不是死守默认值。默认高水位64KB针对的是普通业务当你做的是高吞吐的大消息传输时很容易出现网络瓶颈未到、水位线先触发的情况导致业务侧过度退避。反过来如果做的是大量低延迟小消息推送默认水位线又太宽松内存积压风险变大。合理的做法是在启动时根据业务流量模型和缓冲区平均占用情况预估一个合适的WriteBufferWaterMark值然后配合监控逐步微调。第三写数据要注意背压与批量的平衡。频繁flush可以提高实时性但也会增加系统调用次数攒一批再flush能降低系统调用开销却会提高延迟。没有万能方案只能是针对具体业务做取舍。我的经验做法是如果对端是下游服务而不是用户终端完全可以采用攒批flush的策略延迟多几十毫秒通常无感如果对端是用户终端比如推送系统那更合适的通常是半攒批策略比如按时间窗口每5ms或10ms批量flush一次既保证吞吐又有可接受的延迟。6.3 一个值得深挖的细节write时主动设置过期时间在这个系列的所有源码分析里写数据链路是唯一一个让我觉得看起来简单、实则暗藏玄机的部分。最后分享一个冷门但有用的技巧当你给write设置ChannelPromise时可以主动通过scheduledExecutor给这个promise加一个超时检测。因为正常情况下如果socket发送缓冲区满doWrite会注册OP_WRITE等待可写这个等待时间是不确定的。如果没有超时机制业务线程可能会一直在future.await上挂很久。我遇到过一个极端情况对端进程僵死socket既不关闭也不读数据我们服务端的消息积压在发送缓冲区所有writeAndFlush的future永远不完成最终业务线程池被全部占满服务彻底假死。如果当时在promise上挂了超时检测就能及时发现对端消费异常触发丢消息保护或者断连重连不至于让整个服务被拖垮。具体实现不复杂在发送时往EventLoop的任务调度中提交一个定时任务超时后判断promise是否已完成未完成则主动fail。这么做有个前提日志里必须有对应的告警通道否则超时fail也只是静默处理。写数据的源码看到这里这条链路的基本骨架已经足够清晰了。从一开始的write入队到flush触发底层写系统调用再到背压机制和失败兜底每一环都有它存在的必要。实际操作中理解源码的最终目的是让我们做出更正确的业务决策——知道什么时候该批量写、什么时候该限流、什么时候该断连只有把这些细节吃透才能真正驾驭Netty的性能。