做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：发送操作的包装对象，一条待发送消息在客户端内部的"身份证"。MessageImpl：Pulsar Message的具体实现，包含消息数据、Key、Properties和构建时生成的MessageMetadata。BatchMessageContainerImpl：批处理器，小消息在缓冲区里合并。MemoryLimitController：客户端全局内存配额控制器，防止异步发送把堆内存打爆。ClientCnx：一个TCP连接对应的Channel处理器，负责把Pulsar协议指令编码并写入Socket。PulsarClientImpl：维护连接池、内存控制器和全局线程池。
这些类都在pulsar-client包下面，包路径比较长，建议直接用IDE打开源码工程，按类名搜索是最快的。
1.3 一个最小发送入口和它背后的状态迁移
举个最简单的例子，后文所有源码都围绕这个入口展开：
Producer<String> 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的入参是Message<T>，但实际上用户构造的是一条"逻辑消息"，它可能没有经过压缩、没有序列化完整，一切都要等到真正进入发送阶段才定稿。这也是异步客户端常见的设计：尽可能延迟不可变对象的创建，减少在高频调用路径上的无用开销。
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一直不返回Receipt，pendingQueue就会一直占用，最终触发超时重发或者背压。所以监控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默认true），ProducerImpl在拿到一条消息后并不会立刻执行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写ByteBuf，Netty线程怎么和Broker握手
5.1 CommandSend怎么组装，payload如何与协议帧拼接
当消息确定要发送（不管是单条还是batch），ProducerImpl会创建一个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流里都是一个独立的frame，batch模式下多个子消息会被压缩到一个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迟迟不回ack，close会等到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 参数配比的实践经验
最后分享一下我实际生产中比较顺手的组合：小消息高吞吐场景，开启batch，BatchingMaxMessages保持默认1000，BatchingMaxPublishDelay调到2-5ms，maxPendingMessages提高到5000，同时把sendTimeout从默认30s下调到10s，换取更快的失败感知。超大消息场景，关闭batch，开启chunking，同时把memoryLimit适当调高，但要和JVM堆内存之间留出足够余量，避免内存控制器没触发反而先OOM。
每个环境的数据特征不一样，这些数字未必照抄，但排查思路是通用的：先确认pending是否增长，再看内存预算是否耗尽，最后检查broker ack延迟和网络重发次数。按这条链路一层层看下去，大部分发送端问题都能在自己客户端源码或者监控指标里定位到，而不需要抓包猜谜。如果只想记住一个结论，那就是——sendAsync返回的MessageId只是一个本地临时值，真正让别人相信"消息已发送"，一定是在拿到Broker的Receipt之后。
