Pulsar 的 Producer 源码我以前读过好几遍每次都是带着线上问题去查的。最开始接手维护 Pulsar 集群的时候最头疼的就是用户反馈“消息发不出去”“发送时延抖动厉害”“生产者只重试不成功”这些问题如果只看 Broker 日志、监控指标往往只能看到冰山一角。真正要定位根因必须把客户端生产者的消息发送链路从源码层面理清楚。这篇文章我把生产者的消息发送底层实现完整拆解一遍从你调用send()那一刻开始到数据包真正写到 TCP 连接上每一步都讲清楚你之后不管是调参、查问题还是做二次开发都能有一个全局视角。开头先划个范围这篇文章聚焦 Java 客户端版本基于 2.10.x 左右的稳定分支。这一版里生产者的核心实现类已经相当成熟跟后续 2.11、3.x 相比变化不大理解这一版之后去看新版源码基本可以直接平迁思路。1. 生产者核心架构从 ProducerBuilder 到分区生产者1.1 创建阶段ProducerBuilder 只是壳ProducerImpl 才是灵魂很多人在看 Pulsar 客户端源码时习惯先从PulsarClient.newProducer()看起然后顺着链条去追create()方法。这个思路没错但容易迷失在大量的 Builder 参数校验和可选项配置里。我建议你换一个角度先抓住创建结果——一个ProducerImpl实例或者更准确地说一个PartitionedProducerImpl实例然后再反向理解 Builder 中那些参数到底是在配置什么。当你创建生产者时如果目标 topic 是分区 topic客户端并不会直接创建一个ProducerImpl而是创建一个PartitionedProducerImpl。这层封装很重要它内部会做一件事发起一次 lookup 请求从 Broker 获取这个 topic 的分区数量partitions 数然后为每一个分区分别创建独立的ProducerImpl实例。也就是说如果你有一个 3 分区的 topic那么你的生产者客户端底层实际存在 3 个独立的生产者对象它们各自维护自己的队列、批处理容器、发送线程和消息确认状态。为什么这样设计直接用一个生产者在内部做路由不好吗不行。Pulsar 的分区语义是每个分区都是一个独立的 append 日志消息的有序性保证只在单个分区内有效。如果共用一个底层连接和队列分区间隔离性会变差一个分区阻塞消息可能拖累其他分区的发送。独立ProducerImpl之后每个分区可以独立控制内存占用、独立重试、独立统计这为后面很多精细化调优留下了空间。这个设计带来的直接后果是你在 Producer 级别的配置比如maxPendingMessages、blockIfQueueFull其实会被分到每个分区生产者上单独生效而不是整个 Producer 共享一个额度。这是很多人调参时踩的第一个坑后面我会专门讲。1.2 生产者状态机还没发消息状态管理已经在工作了ProducerImpl内部维护了一个线程安全的状态机状态定义在ProducerState枚举里Init刚创建尚未完成注册Connecting与 Broker 连接建立中正在注册生产者Ready注册成功可以正常发送消息Close已关闭Terminatedtopic 已被终结或删除生产者不可用发送消息前producerImpl.sendAsync()会检查当前状态只有Ready状态才允许消息进入发送队列。其他状态的处理不同Connecting状态会把消息放入 pending 队列等待连接恢复Terminated或Close直接返回异常不会让你把消息发出去。这点很关键因为你在生产环境里看到“发送超时”的错误有时并不是网络问题而是生产者状态已经异常。最常见的场景是topic 被卸载unload或者 Broker 发生主备切换客户端的 producer 会从Ready切到Connecting。此时消息如果继续进来并不会直接报错而是堆积在内存队列里——如果堆积时间超过你的发送超时时间才会以超时异常的形式暴露出来。所以排查发送超时的时候先看一眼生产者的状态变化记录比无头绪地调超时参数高效得多。源码里有一个很典型的实现sendAsync不是直接发消息而是先调用canEnqueueRequest做一系列判断再通过producerStats记录统计数据最后把消息交给pendingQueue或批处理容器。这一步的严谨程度决定了你在高并发下能不能及时发现异常。2. 消息发送链路逐层拆解你的一行 send()背后经历了什么2.1 从 sendAsync 到分区选择消息带着什么信息进入队列用户调用producer.newMessage().value(bytes).send()最终会落到sendAsync()。如果你跟着源码走你会发现ProducerImpl.sendAsync()内部经过多层包装用户的Message?会被封装成MessageImpl然后判断 topic 是否分区分区 topic 走到PartitionedProducerImpl单分区 topic 直接进ProducerImpl在分区场景里PartitionedProducerImpl会调用MessageRouter来选择一个分区MessageRouter有几个内置实现单分区路由、轮询路由、哈希路由。默认是轮询还是哈希取决于你创建 producer 时是否设置了 key。如果你给消息设置了 key会走哈希路由没设置 key默认轮询。这里想提醒一点哈希路由用的是 key 的 hash 值对分区数取模如果你需要严格保证相同 key 的消息有序必须显式设置 key并且保证分区数不变。分区数变更会对既有 hash 分布造成冲击这也是运维时要注意的。选完分区之后消息才真正进入对应分区ProducerImpl的发送流程。到这一步为止还没有任何网络 I/O一切都在内存里完成。2.2 内存队列与背压容量控制是核心ProducerImpl内部有一个非常关键的组件PendingSendQueue部分版本里叫PendingMessagesQueue。这个队列存储的并不是裸的MessageImpl而是一个个OpSendMsg对象。先解释一下OpSendMsg它把一条或多条消息、消息的字节数据ByteBuf、发送时间戳、回调函数、序列号sequenceId打包在一起作为一次发送操作的最小单元。这么做的好处是回调、重试、确认acknowledge都能以“操作”为单位做管理不至于在回调层面纠缠消息粒度的细节。队列容量受两个参数控制maxPendingMessages队列中最多能堆积多少个OpSendMsg实际是消息数或批次数的累加maxPendingMessagesAcrossPartitions分区生产者间共享的全局容量当队列塞满时blockIfQueueFull决定行为true调用线程阻塞等待队列有空间false直接抛出ProducerQueueIsFullError异常这个设计就是典型的背压机制。很多消息中间件的客户端生产者都有类似结构但 Pulsar 的特点在于队列积压不仅影响延迟还会直接影响后续消息的批处理效果。可以把pendingQueue理解成一个水库水库水位低水流平稳水库满了上游就会被堵住。如果你发现生产者常报ProducerQueueIsFullError不要一心只调大maxPendingMessages也要看看下游消费是否正常、Broker 是否出现写入慢的问题——背压是在保护你不是在惩罚你。这里插一个实际经验曾有一个业务团队把maxPendingMessages调到 100 万结果一个分区 Broker 短暂抖动瞬间堆积几十万条消息在外内存里直接触发了客户端进程 GC 压力飙升。内存保护参数调太大等于亲手拆掉了安全阀不推荐这么干。2.3 消息在批处理容器里等什么时序图思维下的发送触发器在非批量模式下一条消息进入队列会立即触发发送。但在默认开启批处理enableBatchingtrue时消息并不会立即发送而是先进入一个BatchMessageContainer。BatchMessageContainer是另一个核心组件它把多条消息粘合成一条CommandSend请求。那什么条件下这个容器里的消息会被“倒出去”发送源码中主要在三个触发器里做判断消息数达到batchMaxMessagesNum默认值 1000累加字节数达到batchMaxBytes默认 128KB 或 1MB 不等取决于版本和配置定时器到期sendTimeout默认 30s和 batch 相关的定时轮询周期需要一个很直观的理解BatchMessageContainer就像一个公交车人满即走或者到了发车时间就走。这样能显著减少发送请求的数量从而提高吞吐。但是这里有个隐藏的坑批处理会引入额外的延迟。如果你的业务场景是低频发送、但要求单条消息低延迟那默认开启批处理反而可能让你每条消息多等几十甚至上百毫秒。源码中批处理的 flush 动作在线程池里有个batchTimer在驱动它会周期性检查每个容器是否有需要超时刷出的消息——周期通常是 1ms 到 100ms 之间所以你可以感知到开启批处理后单条消息延迟至少会有这几个毫秒的保底开销。我在另一个团队见过一个经典误用他们把消息发送频率很低每秒几条却开着批处理还用了默认参数结果每条消息平均延迟比不开启批处理高了一个数量级。后来把batchingEnabled关掉或者把batchingMaxPublishDelay调低延迟立刻恢复了。记住批处理是给高吞吐设计用的不是给低延迟场景用的。2.4 网络层的最后一公里ClientCnx 如何把请求写到 Socket消息批次最终会从容器中取出序列化成协议包然后交给ClientCnx写入 Netty Channel。ClientCnx是 Pulsar 客户端连接层的核心它本质上是一个 NettyChannelHandler。看源码时你会注意到它内部持有一个PendingLookupRequests、PendingRequests等映射结构用于在客户端等待 Broker 响应时做请求-响应的匹配。发送消息时OpSendMsg里已经准备好了要发送的ByteBuf这个ByteBuf是对协议数据的内存封装不是byte[]ClientCnx直接调用channel.writeAndFlush(byteBuf)。这一步就是真实地封装成 TCP 数据包发出去。但在此之前还有一个重要动作因为批处理发送的可能是一个CommandSend里带有多条消息的协议格式编解码逻辑在Commands工具类里。这个类是整个协议层的枢纽它生成各种协议的 ByteBufConnect、Lookup、Producer、Send、SendReceipt、PingPong 等。如果你要自己实现一个 Pulsar 客户端Commands类就是最值得花时间研究的部分。网络层面的另一个点是 TCP 连接复用和连接池。PulsarClientImpl底层维护了一组到各 Broker 的连接基于ConnectionPool一个生产者使用哪个连接取决于它注册到了哪个 Broker。分区 topic 的各个分区可能分布在不同的 Broker 上所以一个PartitionedProducerImpl实际管理的各个ProducerImpl底层连接的目标可能各不相同。这也就解释了为什么分区生产者特别容易在某个分区 Broker 出问题时只出现局部消息发送异常——你的生产者在宏观上是分片独立的。3. 核心机制与关键代码逻辑批量发送的终极形态与确认机制3.1 批量消息容器合并的收益和内存代价BatchMessageContainer有两个实现一个是适用于非压缩场景的实现一个是用于压缩场景的BatchMessageContainerImpl不同版本实现名称可能略有差别。它们共同的核心接口是BatchMessageContainer关心这几个方法addMessage(MessageImpl?, SendCallback)往容器里加消息havePendingMessages()容器里有没有等待发送的消息getBatchSizeBytes()累计字节数createBatch()/discard()生成一个批次或丢弃当前所有消息看一下addMessage的逻辑每加入一条消息容器要决定是否立即触发flush。它会对比batchMaxMessagesNum和batchMaxBytes。一旦达到阈值就设置标志位让上层发送。这里有个很多人忽略的点消息在addMessage时不会立刻复制数据而是保留原消息的引用等到 flush 时统一做大数组拼接或压缩编码。这种写时合并的思路能有效降低内存拷贝次数。但代价是什么一是内存占用更高因为保留的是多个消息对象引用二是如果容器 flush 失败重试时这些消息是作为一个整体重发的——也就是说重试粒度变大了。想象一下一辆公交车满载 1000 人走到半路抛锚整车人再重新上一辆新车整体重传会增加延迟放大这是批处理天然的特性。所以开启批处理后如果你的 Broker 偶发抖动客户端这边感受到的发送超时往往是一批一批的超时而不是单条消息超时。3.2 确认机制从 SendReceipt 到回调触发消息发送到 Broker 后Broker 会返回CommandSendReceipt里面包含producerId、sequenceId、messageId。ClientCnx收到这个回包后会根据producerId sequenceId找到之前缓存的OpSendMsg然后调用callback.completed()完成一次发送的闭环。这可能返回的是MessageId那类对象或者是你自己注册的SendCallback。这里有两点值得深挖第一点sequenceId是生产者本地的自增 ID客户端用它做请求匹配。因为消息可能乱序到达极端网络场景下客户端必须靠sequenceId而不是发送顺序来匹配回执。源码中确认到达的处理逻辑在ProducerImpl.ackReceived()里它会从队列中找出对应OpSendMsg移除并更新lastSequenceIdPublished。第二点如果消息发送超时TimeoutHandler在ClientCnx的定时任务里会检查 pending 队列里OpSendMsg的时间戳。超过sendTimeout的请求会被标记为发送失败触发callback.exception()。但默认实现里生产者不会自动重发一条已经标记失败的消息重试逻辑要么由你在回调里实现要么依赖消息自身的生命周期比如通过生产者重试策略。这一点在 Pulsar 里和 Kafka 很不一样Kafka 的send()自带重试参数Pulsar 的sendAsync并不默认重试。很多从 Kafka 迁过来的团队最容易在这里产生误解。3.3 消息压缩的实现时点容器 flush 之后、写网络之前在开启压缩比如 LZ4、ZSTD、Snappy时压缩动作发生在哪一步不是每条消息单独压缩而是在一个 batch 生成后对整个 batch 的消息体做统一压缩。看源码时你会找到这样一个调用链BatchMessageContainerImpl.flush()→ 把所有消息的 payload 拼接成一个ByteBuf→ 调用CompressionCodec.encode()→ 得到压缩后的ByteBuf→ 作为 batch 的 payload 写入Commands.newSend()。这个“先拼后压”的设计很有讲究如果每条消息单独压缩压缩率会很低尤其当单条消息体很小比如几十字节的日志时压缩头都占了不少比例。而批处理 统一压缩可以拿到更高的压缩比。代价是压缩过程会消耗一次额外的 CPU 开销并且 Broker 侧需要对压缩后的 batch 做解压才能消费单条消息所以压缩不只是在客户端有成本。有一个非常常见的问题启用了压缩但吞吐没有提升甚至下降了。如果你压的是图片、视频等已压缩内容或者消息体本身就很小且数量不多压缩 CPU 开销可能会大于传输节省。建议保留默认压缩参数并根据消息类型做实测。3.4 分块发送超过 maxMessageSize 的消息如何处理默认单条消息的最大允许大小是 5MBmaxMessageSize客户端会在消息进入容器前做一个检查。如果你要发送超过这个限制的消息有两个方向调大 Broker 的maxMessageSize配置同时客户端侧allowChunking控制开启分块发送chunking客户端把大消息拆成多个分块按顺序分开发送Broker 收到后按消息头里的 chunk 信息重新组装分块在源码里对应ChunkMessageIdImpl和分块消息元数据。实现上分块不是简单地把 body 切开而是会在消息元数据里加入 chunk id、chunk 总数、原始消息长度等信息。每个分块拥有独立的 messageIdBroker 在收到完整分块序列后组装为真正的消息并持久化。分块发送的坑在于分块会跟批处理产生冲突源码里如果启用了 chunking会强制关闭该条消息的批处理因为分块消息占的字节数大跟其他消息混在一个批次里会导致组装逻辑复杂度剧增分块发送时每个分块都可能被不同 Broker 接收吗不一定。正常情况下同一个 Producer 发出的分块会顺序发送到同一个分区但如果 topic 发生重平衡或连接重连下游能否正确重组取决于这个 topic 的持久化语义分块消息在 Broker 端重组如果其中某个分块在传输中丢失比如连接断开、超时整条大消息只能重发。所以网络抖动场景下大消息发送成功率反而可能低于小消息大消息的发送我一般建议超 5MB 就优先评估是否走对象存储、把消息体换成引用或 URL而不是硬着头皮开 chunking。开了 chunking你的网络健康度要求会提高一整档。4. 生产环境常见问题与排查技巧4.1 发送延迟突然升高怎么定位先把链路拆成两段考虑客户端内存队列积压和网络到 Broker 耗时。客户端内存队列积压可以从生产者监控指标里看pendingQueueSize是否在上涨或者batchMessageContainer中滞留消息数是否异常。如果队列积压持续存在说明发送速度小于消息产生速度或 Broker 端写入出现阻塞。这时候再去查 Broker 的 journal 写入耗时、BookKeeper 的 fsync 耗时反向确认瓶颈。如果是网络耗时增加ClientCnx的 TCP 连接 RTT 和丢包率会更直观。注意 Pulsar 客户端内部有 Pulsar 的ClientCnxn层底层完全依赖 Netty。打开 Netty 的流量记录能看到连接是否有频繁的writeAndFlush失败或重连事件。经验技巧给生产者增加producerName的标识在日志采集时带上 topic 名和应用名问题是“哪个分区延迟高”还是一目了然。很多团队不区分分区结果一个问题定位半天。4.2 ProducerQueueIsFullError 不断出现怎么处理出现这个问题只有三个根因消息产生速度超过了写入速度、Broker 侧故障恢复慢、配置不合理。第一步确认是哪个分区的问题看 topic 的每个 partition 是不是都有积压。如果不均匀优先处理热点分区。第二步看每个ProducerImpl的pendingQueue数量和memoryLimitController的使用量。第三步把blockIfQueueFull设为true缓解客户端异常爆发但注意线程会被阻塞服务能力下降。如果业务本身允许丢弃建议在修复前暂停部分生产流量而不是无限扩大队列。扩大队列不是解决速度差的根本办法它只会把故障延迟暴露的时间拉长。你没看错这里就是那个“队列大小”参数陷阱的高发区。官方默认maxPendingMessages是 1000maxPendingMessagesAcrossPartitions是 50000这对大多数低延迟业务是够了。如果你调得过大遇到 Broker 故障时堆积在外内存里的消息可能让 JVM 直接 OOM。4.3 消息乱序批处理和重试如何影响顺序Pulsar 保证分区内有序前提是“单生产者”场景。如果你有多个生产者实例写同一个分区序只能保证到消息持久化的顺序不由某个客户端单独控制。在单生产者场景批处理模式下乱序一般不会发生因为 client 是严格按 sequenceId 顺序发送批次。但有一种例外开启失败重试后后一个 batch 可能先发成功前一个 batch 重试后发送这就会导致最终持久化顺序与原始消息顺序不一致。源码里OpSendMsg的重试发送并不是按照原 sequenceId 顺序严格排序的它只是重新把数据放到相应的ProducerImpl内部队列中而队列如何调度是由时机决定的。所以如果你的业务严格要求顺序建议关闭批处理enableBatchingfalse让每条消息独立发送天然按顺序发送关闭生产者的自动重试把重试逻辑上移到业务层保证重试时顺序可控使用newMessage().orderingKey(...)控制分区内顺序本质还是控制路由4.4 常见问题速查表现象直接原因排查路径建议处理发送延迟高队列积压/网络抖动pendingQueueSize指标、Broker 写入耗时调大sendTimeout并定位瓶颈别盲目加队列ProducerQueueIsFullError下游写入速度跟不上内存计数、分区热 Key检查订阅消费情况降级或限流必要时扩容发送超时频繁Broker 主备切换/unload看客户端状态机是否在 Connecting等待重连减少瞬时生产流量大批量消息发送失败超maxMessageSize查看消息体大小评估分块或改存对象存储压测时吞吐不达标批处理参数/压缩参数不当对比开/关批处理和压缩针对场景做参数矩阵实验消息乱序多生产者同写/重试插队确认生产端拓扑、重试次数单生产者 关闭重试或业务层排序我给一个真实案例有一次线上压测发现压测机 CPU 使用率高但吞吐只有预期一半。排查后发现是开启了 ZSTD 压缩而压测数据是随机字节完全压不动。关闭压缩后吞吐线性上涨。所以压测要用真实业务消息样本不要拿随机字节凑数。5. 源码阅读路径建议从这些关键类入手如果你想自己深入读一遍生产者的发送链路我建议按这个顺序读PulsarClientImpl—— 初始化和连接池入口ProducerBuilderImpl—— 参数解析的终点ProducerImpl—— 核心状态机和发送入口PartitionedProducerImpl—— 分区路由和子生产者管理BatchMessageContainer/BatchMessageContainerImpl—— 批量语义的核心OpSendMsg—— 发送单元的真正组成ClientCnx—— 网络协议层Commands—— 协议包的字节级编解码看这些类的时候先不看细节先把每个对象的生命周期捋一遍谁创建了谁谁的引用被存到了哪里在超时、失败、重连时这些引用如何处理如果每个问题都能在三层以内回答生产者这块你就真正吃透了。之后再去看PulsarAdmin、Consumer、Reader等模块会发现同源设计事半功倍。还有一条经验Pulsar 客户端源码中大量用了异步回调和 CompletableFuture读的时候别急着弄清楚每个 future 的完成路径先关注回调在哪里触发、异常在哪里捕获。发送链路中的异常捕获点往往就是问题定位的咽喉。比如sendAsync失败时异常会被包在SendCallback里消息确认超时时在TimeoutHandler连接断开时在ConnectionHandler。把这三个位置记住排查线上问题能少走一半弯路。最后再分享一个小技巧调试生产者发送问题时在客户端日志开启Debug级别Pulsar 客户端会输出发送回执的同步日志。虽然生产环境不建议开着但在测试环境复现问题的时候这一招比看任何反编译工具都直接。配合tcpdump抓包观察协议交互过程你甚至能直观看到一条消息从CommandSend到CommandSendReceipt的完整网络往返。源码解析结合抓包验证是我个人认为理解消息中间件客户端最有效的一对组合拳。