做Pulsar生产者的二次开发或者排查发送端问题的时候大部分人都会卡在同一个地方sendAsync()明明返回了MessageId但这条消息到底走到哪一步了、Broker有没有真的收到心里完全没底。因为Pulsar客户端和Broker之间是异步通信一条消息要经过内存水位确认、队列排队、batch聚合、TCP发送、Broker落盘确认这一整条链路最后回调才会触发。这篇文章直接把Apache Pulsar客户端生产者消息发送的底层源码链路拆开从send()入口一路看到Netty写Socket把关键对象、线程切换、内存控制、批处理分块和超时重发机制讲清楚。适合想读Pulsar源码但不知道从哪下手的读者也适合正在排查发送超时、队列积压、发送端内存占用异常的开发者。1. 先理一遍发送链路一次send()要穿越几道关卡才会真的发出去1.1 链路其实是两段异步API背后的队列和解耦首先要打破一个直觉Pulsar的Producer并不像传统同步Socket那样调一次write就直接把字节推给Broker。sendAsync()在调用线程里做的事情其实很少大部分逻辑都被异步编排到了Pulsar内部的调度线程上。整条链路可以拆成两段第一段是从用户线程进入ProducerImpl完成消息封装、参数校验把一条消息塞进pendingQueue。第二段才是真正的发送由内部线程从pendingQueue里取出OpSendMsg构造协议指令通过Netty写进连接。这就是为什么ProducerImpl里大量逻辑都是提交Runnable而不是直接执行。第二段进一步拆分可以看到更细的关卡先要检查内存水位MemoryLimitController是否还有预算然后看当前消息是否满足批处理条件能进batch container就先聚合不能聚合就直接生成CommandSend紧接着编码成NettyByteBuf在ClientCnx上writeAndFlush最后等Broker回CommandSendReceipt再把callback从pendingQueue里摘出来。每一步之间都是异步衔接读源码的时候千万不要在一个方法里寻找连续调用栈要跟着回调走。1.2 核心类的分工先记住这些对象再进入细节这里列出这次解析会反复出现的类先记住它们读代码的时候就不容易迷路ProducerImpl生产者的核心实现负责队列管理、背压控制、消息封装、序列号分配。OpSendMsg发送操作的包装对象一条待发送消息在客户端内部的身份证。MessageImplPulsar Message的具体实现包含消息数据、Key、Properties和构建时生成的MessageMetadata。BatchMessageContainerImpl批处理器小消息在缓冲区里合并。MemoryLimitController客户端全局内存配额控制器防止异步发送把堆内存打爆。ClientCnx一个TCP连接对应的Channel处理器负责把Pulsar协议指令编码并写入Socket。PulsarClientImpl维护连接池、内存控制器和全局线程池。这些类都在pulsar-client包下面包路径比较长建议直接用IDE打开源码工程按类名搜索是最快的。1.3 一个最小发送入口和它背后的状态迁移举个最简单的例子后文所有源码都围绕这个入口展开ProducerString producer client.newProducer() .topic(persistent://public/default/trace-topic) .sendTimeout(30, TimeUnit.SECONDS) .enableBatching(true) .enableChunking(false) .create(); producer.newMessage() .key(order-1001) .value(Hello Pulsar.getBytes(StandardCharsets.UTF_8)) .sendAsync() .thenAccept(messageId - log.info(sent: {}, messageId));这条消息从提交到回调触发会依次经历消息封装 - 序列号分配 - 加入pendingQueue - 内存预算申请 - batch判断 - 协议编码 - TCP write - broker持久化 - SendReceipt回来 - pendingQueue移除 - 用户回调执行。整个生命周期里发送成功只有一次就是收到Broker的Receipt那一刻。理解了这条主干后面所有源码细节都是在这条链路上做文章。2. sendAsync()里的三道闸门封装、序列号、消息大小2.1 MessageImpl是怎么构建的底层二进制长什么样进入ProducerImpl之后第一步就是构造MessageImpl。newMessage()返回一个MessageBuilder调用value()会把用户对象序列化字节数据保存在Payload中调用key()会把字符串Key放进metadata的partition_key字段。这里有一个让很多人意外的点MessageImpl在发送前并没有完整拼好协议报文它只是把元数据和payload分开保存真正的二进制组装发生在网络编码阶段。MessageMetadata是protobuf结构关键字段包括producer_name、sequence_id、publish_time、partition_key、uncompressed_size、compressed_size、compression等。消息封装完成后会立刻在MessageImpl里预分配一个本地MessageIdImpl。注意此时生成的MessageId只是客户端本地估计的ID不是最终Broker确认的MessageId。源码里这个ID会在回调future中先放一份最终会被Broker返回值正确填充。读这块代码时你会看到sendAsync的入参是MessageT但实际上用户构造的是一条逻辑消息它可能没有经过压缩、没有序列化完整一切都要等到真正进入发送阶段才定稿。这也是异步客户端常见的设计尽可能延迟不可变对象的创建减少在高频调用路径上的无用开销。2.2 sequenceId的分配以及它在去重里的角色每个Producer在创建时如果没指定initialSequenceId客户端会用0作为起点如果指定了就从这个值之后开始递增。每一条进入发送流程的消息都会在sendAsync路径上通过一个原子自增拿到一个单调递增的sequenceId写入MessageMetadata。这个序号有两个作用一是让Broker可以识别同一Producer的消息顺序二是配合Broker端的去重机制同一序号的消息如果出现两次Broker会丢弃重复的那条。实现细节上ProducerImpl维护了lastSequenceId和序列号分配器分配采用的是CAS操作而不是全局锁避免高吞吐下序列号分配成为瓶颈。这里要提醒一下sequenceId只跟Producer实例绑定无法通过修改消息Key或者其他元数据来改变如果你在回调里去调整initialSequenceId只会影响后续新建的消息已经入队的消息不受影响。出于工程严谨性sequenceId的生成和OpSendMsg的入队是同一个临界区内完成的保证分配了序号但消息没入队的记录不会出现。这在乱序排查中很重要如果你在看监控时发现sequenceId跳号多数情况下不是因为客户端漏发而是因为中途有消息入队失败被拒绝了。2.3 消息大小边界超过maxMessageSize会发生什么sendAsync里还有一道极易忽略的校验消息太大。客户端在连接建立后从Broker获取maxMessageSize默认通常是5MB。当单条消息字节数超过这个上限有两种走向如果开启了chunking消息会被切片分段发送如果没有开启则直接抛出异常或者通过SendCallback返回失败。源码里的判断位置在消息封装之后、pendingQueue入队之前也就是说超长消息根本没有机会进入排队环节。maxMessageSize是客户端与服务端协商的结果不是纯粹本地配置。这意味着一条消息在这个时刻合法换一个Broker参数就可能非法。如果你在生产环境调整过Broker的maxMessageSize客户端必须重连才能拿到新值这是一处容易导致为什么改完配置还报错的坑。我遇到过一次线上大消息偶发失败调到客户端那边看日志发现客户端还在用旧的5MB阈值而Broker已经把限制改成4MB了重连之后问题立刻消失。3. pendingQueue与背压设计为什么拿到就发在这里行不通3.1 OpSendMsg在等待receipt期间消息存在哪里sendAsync之后消息并不会直接进入Netty而是先包装成一个OpSendMsg对象放进ProducerImpl的pendingQueue。等待队列的这个设计其实是在刻意拉开网络I/O和业务提交的速度差Broker落盘需要时间网络往返有延迟如果每提交一条就同步等ack生产者吞吐会被压在单线程往返延迟上。OpSendMsg里保存的不只是消息本身还有发送时的sequenceId、业务回调SendCallback、消息创建时间createdAt、以及当前重发次数。pendingQueue实际上是一个有界队列队列里的每一个元素都代表客户端已经提交、但还没拿到Broker ack的请求。生产上如果看到pending数量持续上涨基本可以断定Broker侧消费或者持久化出了问题而不是客户端自己的逻辑卡住了。这里再解释一个容易误解的现象writeAndFlush成功之后消息依然在pendingQueue里。因为TCP的发送成功只代表字节进了内核缓冲区不代表Broker已经处理完毕。如果Broker一直不返回ReceiptpendingQueue就会一直占用最终触发超时重发或者背压。所以监控pendingQueue的长度比Netty的write状态更接近真实发送状态。3.2 maxPendingMessages与信号量有界队列怎么实现pendingQueue内部是一个ArrayDeque但光有队列还不够因为入队操作可能瞬间把内存打满。ProducerImpl内部还有一个Semaphore许可数量就是maxPendingMessages默认1000。发送前必须tryAcquire成功否则就会进入backpressure分支一旦Broker ack回来release一个许可队列和信号量同时腾出空间。读源码的时候注意一个细节信号量和队列的入队顺序是有讲究的。源码会先成功获取许可再入队避免出现队列有元素但信号量空着的不一致状态。反向看如果许可获取不到sendAsync会走哪个分支取决于blockIfQueueFull参数。默认情况下这个参数是false队列满时新的sendAsync会通过回调返回异常改成true时提交线程会阻塞等待直到队列腾出位置。这个阻塞是用户线程直接阻塞不是异步等待所以生产上开blockIfQueueFull一定要谨慎配置过高容易把业务线程全部卡死在发送上。3.3 blockIfQueueFull到底挡了什么挡不住什么需要澄清一个容易混淆的地方blockIfQueueFull阻塞的是用户线程提交而不是Broker确认。也就是说它只对发送端起到限流作用消息一旦进了队列后面网络层慢不慢、Broker ack快不快它都管不了。生产经验上如果发送端QPS波动很大我宁可让发送返回失败走业务重试也不要让所有线程阻塞在等待队列腾空上尤其是下游Broker出现故障时阻塞等待往往会变成线程池耗尽。不过有一种场景blockIfQueueFull很有用短时流量尖峰且Broker只是暂时变慢把提交线程短暂卡住能避免大量消息被丢弃。这种情况下最好配合一个较短的发送超时避免卡死时间过长。我的建议是先确认你的业务是否能容忍发送失败能容忍就走默认模式不能容忍再考虑阻塞模式同时给每台客户端设置maxPendingMessages上限防止内存被打满。3.4 MemoryLimitController整个客户端共享的内存水位pendingQueue控制的是每个Producer的悬挂请求数量MemoryLimitController控制的是整个客户端进程允许被消息占用的大块堆内存。Pulsar客户端在创建时默认给这个控制器配置64MB上限每次消息进入发送流程时会通过tryAcquire向控制器申请内存消息完成无论成功还是失败后会release。当内存总量逼近上限时新的发送就要等待等待逻辑和信号量是两套独立机制。这个设计非常重要因为Pulsar客户端是异步模型如果对每条消息占用的ByteBuf不做全局预算只要生产者一多堆内内存就会失控。读源码时建议重点看MemoryLimitController.tryAcquire的实现它内部维护了currentUsage和maxMemory用volatile修改变量配合CAS控制并发安全没有用重量级锁。这种做法的好处是高吞吐场景下内存预算的申请成本几乎可以忽略坏处是极端并发下可能存在短暂超卖但整体上不会危害进程稳定性。4. 批处理与分块两个让一条消息变身多条/多段的机制4.1 batch的核心容器聚合并批量发送如果开启了批处理enableBatching默认trueProducerImpl在拿到一条消息后并不会立刻执行doSend而是把它塞进BatchMessageContainerImpl。这个容器会先积累消息等满足批量条件后再一次性把它们打包成一条批量消息发送。对Broker而言一条批量消息是携带多条子消息的单个协议报文既减少了网络往返也减少了单条消息的持久化开销。这里涉及三个关键参数BatchingMaxMessages默认1000条、BatchingMaxBytes默认128KB、BatchingMaxPublishDelay默认1ms。三个条件满足任意一个容器就会触发flush。源码里的判断逻辑在flushIfNeeded方法中每次新增消息都会检查这三个阈值。如果你只需要微调其中一个其他参数要保持默认值否则可能把容器推向另外一个方向。4.2 flush触发时机什么是压缩率最高的窗口真正理解batch要把握flush时机。假设业务单条消息很小比如几十字节如果把BatchingMaxPublishDelay调到10ms那10ms内所有消息会被聚合成更大的包吞吐会明显提升但每条消息的可见延迟会接近10ms。源码里batch容器在flush时会把容器里的多条消息编码成一条临时消息并且所有子消息共用一个batch sequenceId。注意batch模式下用户在回调里拿到的MessageId其实是这条批量消息的MessageId如果你需要按单条消息感知成功或失败batch会把粒度放大。源码里回调分发是逐个执行的也就是说即使批量消息整体成功调用方感知到的是N个回调同时触发而不是一个统一的批量回调。对业务埋点来说这里的时间戳就是回调触发时间而不是BatchMaxPublishDelay的到期时间。4.3 大消息切分chunking是怎么编序号、怎么组包的当单条消息超过maxMessageSize且enableChunking(true)时ProducerImpl会把消息payload按固定大小切成多个chunk。每个chunk会拿到独立的sequenceId元数据里的chunk_id和total_chunks标记当前是第几个片段。Broker收齐所有chunk后会按照chunk_id重组完整消息再交给消费端消费端看到的就是一条完整的原消息不需要感知分块。chunk的顺序依赖sequenceId所以源码中ProducerImpl对chunk分配序列号时是连续的中间不能再插入其他消息否则重组就会错位。这也是为什么chunking往往和batch互斥的原因之一batch会一次性发送多条消息和chunk的序号连续性天然冲突。实际阅读代码时你可以看到doSend里专门分支判断如果消息需要chunk就不会走batch容器而是逐chunk生成独立的OpSendMsg。4.4 一个真实的配置教训为什么chunking和batch同开会踩坑Pulsar在设计上允许两者同时存在但如果你真的同时开启会发现ProducerImpl在doSend里chunk会强制绕过batch容器走独立发送路径。源码里确实做了这样一个分支批处理容器只在消息不需要chunking时工作。所以实践中要么处理大量小消息开batch要么处理超大消息开chunking很少需要同时用到。如果你的业务消息大小分布横跨几个数量级我建议按主题拆分小消息主题开batch大消息主题开chunking而不是在同一个Producer上同时打开两个开关。之前有一个项目消息体从几百字节到几MB都有一开始图省事两个开关全开结果大消息的chunk经常和小消息的batch混在同一个连接上虽然逻辑上不出错但排查问题的时候非常难定位后来拆主题后清晰多了。5. 网络发送与超时重发ClientCnx写ByteBufNetty线程怎么和Broker握手5.1 CommandSend怎么组装payload如何与协议帧拼接当消息确定要发送不管是单条还是batchProducerImpl会创建一个SendRequest里面包含producer_id、sequence_id、以及序列化后的MessageMetadata和payload。然后进入ClientCnx这里才是把对象变成字节的地方sendCommandWithPayload方法会先构造CommandSend的protobuf消息再把CommandSend和payload以frame length command bytes payload bytes的组合写进ChannelBuffer。用Netty的writeAndFlush把字节写出后后续就交给了TCP协议栈和Broker端的网络线程。读这段代码时可以看到Pulsar客户端并没有用额外的序列化框架协议顶层的封装就是protobuf加长度前缀整个消息体很轻这也是Pulsar协议性能不错的原因之一。如果你看抓包会发现每条发送消息在TCP流里都是一个独立的framebatch模式下多个子消息会被压缩到一个frame里字节数明显下降。5.2 Broker ack之前消息在哪、谁在等它从writeAndFlush成功到Broker返回CommandSendReceipt中间隔着网络传输、Broker接受、持久化确认应答。这段时间里消息其实还占着pendingQueue并没有因为写出去就释放。也就是说write成功不等于消息成功网络栈层面只看socket write成功与否它代表不了Broker的确认。源码里receipt的处理在ClientCnx.handleSendReceipt方法解析broker返回的messageId后找到pendingQueue里对应sequenceId的OpSendMsg把最终MessageId写入回调future然后从队列中移除并release信号量。这一步是整个发送链路的终点。读代码时你会发现这里还处理了几个边缘情况比如Broker返回的sequenceId和本地对不上或者消息已经被标记为超时正在重发这些分支都是在cancel之后还可能收到旧确认的时序下做的防御。5.3 sendTimeout到期由谁触发重发sendTimeout默认30秒。源码里ProducerImpl在创建时会启动一个ScheduledExecutor按照sendTimeout周期扫描pendingQueue中的OpSendMsg。当一个消息的createdAt距今超过sendTimeout且仍未被ack它就会被标记为超时然后触发一次重新发送。重发逻辑会根据当前连接状态走doSend如果连接正常会在原连接上重发如果连接已经断开会通过连接池重建连接后重发。同时会把这条消息从pendingQueue里临时移除再放回避免重复回调。在较新版本的源码中这个扫描还会对chunked消息做更细粒度的处理只重发未收齐的分片而不是把整条消息重新发送。这些都是消息可靠性的关键细节。5.4 close和flush生产者关闭时和队列的纠葛生产者在调用close()时如果队列里还有pending消息是等还是不等源码里close分为两步首先关闭新消息入口拒绝后续send随后等待pendingQueue清空直到超时。这就是为什么close可能阻塞很久如果Broker迟迟不回ackclose会等到sendTimeout对应的重发周期结束。如果你希望关闭前把队列里已有消息尽量发出去应该显式调用flush()它会把batch容器当前积累的消息强制发出然后等待队列清空。close默认不等batch里未flush的消息这很容易造成最后几条小消息丢失的错觉。所以线上优雅关闭时先flush再close是更稳妥的顺序。我通常还会在关闭前设置一个较短的超时避免某个Broker故障时服务停机流程被拖死。6. 源码阅读顺序与生产经验哪些参数值得调哪些坑我踩过6.1 建议的源码阅读路径如果想自己跟踪一遍我建议按这个顺序先看ProducerImpl.sendAsync再看MessageImpl构建随后进入OpSendMsg和pendingQueue接着看MemoryLimitController再看BatchMessageContainerImpl最后看ClientCnx.sendCommandWithPayload和handleSendReceipt。这个顺序是从业务入口到网络边界单向推进每一步都能在前一步的基础上叠加新的概念。代码阅读时建议用断点方式验证线程名。在sendAsync入口下一行断点观察调用线程通常是业务线程在doSend内部下一行断点观察线程已经切到了Pulsar的scheduled executor在ClientCnx.writeAndFlush断点线程是Netty EventLoop。看到线程名的切换你对异步模型的理解就基本到位了。这套验证方法比单纯读代码快很多也能帮你建立回调在什么线程执行的直觉。6.2 生产环境里常见的两个发送慢问题第一个常见问题sendAsync回调积压。客户端使用异步回调可回调里如果执行了数据库写入或者远程调用回调线程会被堵住进而影响后续消息的发送确认最终表现为pending持续上涨。源码里SendCallback是在Netty线程或scheduled线程上执行的阻塞它等于阻塞了整个EventLoop。我的做法是回调里只做轻量操作重活丢给业务线程池。第二个常见问题是内存限制触发等待。MemoryLimitController达到上限后发送线程会在tryAcquire上等待表现是发送变慢而不是报错。排查时要先确认是不是有大量消息瞬间涌入或者某一条超大payload长期占着内存没有释放。我会在监控里单独把producer pending queue size和内存预算占用率两路指标一起看比只看TPS准确得多。有一次排查发送延迟TPS一切正常但pending一直在涨最后定位到是一条几MB的消息反复重发把内存预算全部占满。6.3 参数配比的实践经验最后分享一下我实际生产中比较顺手的组合小消息高吞吐场景开启batchBatchingMaxMessages保持默认1000BatchingMaxPublishDelay调到2-5msmaxPendingMessages提高到5000同时把sendTimeout从默认30s下调到10s换取更快的失败感知。超大消息场景关闭batch开启chunking同时把memoryLimit适当调高但要和JVM堆内存之间留出足够余量避免内存控制器没触发反而先OOM。每个环境的数据特征不一样这些数字未必照抄但排查思路是通用的先确认pending是否增长再看内存预算是否耗尽最后检查broker ack延迟和网络重发次数。按这条链路一层层看下去大部分发送端问题都能在自己客户端源码或者监控指标里定位到而不需要抓包猜谜。如果只想记住一个结论那就是——sendAsync返回的MessageId只是一个本地临时值真正让别人相信消息已发送一定是在拿到Broker的Receipt之后。