1. 从一次 send() 说起：生产者的消息旅程

先放结论：Pulsar 客户端生产者发送一条消息，表面上只是一次producer.send(msg)调用，但这背后是一条完整的多级链路——消息不是直接"扔"到网络上的，而是先经过内存缓冲、分区路由、批次聚合，再由独立的内部线程统一投递到 Broker，最后以 Receipt（回执）的形式确认完成。这篇文章不聊概念，直接深入源码，把这条链路拆开看。

// 简单得不能再简单的一段代码 Producer<String> producer = client.newProducer(Schema.STRING) .topic("persistent://public/default/mytopic") .create(); producer.send("hello pulsar");

这行send()背后到底发生了什么？我先把整条链路的参与者列出来，后面逐个环节拆解：

• ProducerImpl：生产者核心实现类，负责消息缓冲、批次管理、回调调度
• Partitioner：分区选择器，决定消息进哪个分区
• MessageImpl：消息的内部表示，包含 metadata 和 payload，真正的网络传输对象
• OpSendMsg：一次发送操作的封装，承载消息、回调、发送时间等状态
• ClientCnx：客户端与 Broker 之间的连接通道，负责协议编解码
• SendReceipt：Broker 返回的写入确认，触发用户回调

整个链路概括成一句话：send() → 分区选择 → 入队缓冲 → 批次封装 → 写入连接 → 协议发送 → Broker 确认 → 回调触发。下面一个一个拆。

2. Producer 的创建与状态机：一切发送的前提

2.1 ProducerImpl 的构建过程

client.newProducer()返回的是ProducerBuilder，真正创建生产者时调用的是ProducerBuilderImpl.createAsync()。它的核心逻辑是：

// ProducerBuilderImpl.createAsync() 简化逻辑 public CompletableFuture<Producer<T>> createAsync() { // 1. 从客户端获取或创建 topic 的 partitioned topic 元数据 return client.getPartitionedTopicMetadata(topic) .thenCompose(metadata -> { // 2. 根据分区数决定创建生产者策略 if (metadata.partitions > 0) { return createPartitionedProducer(metadata); } else { return createSinglePartitionProducer(); } }); }

重点在第一步。getPartitionedTopicMetadata()会向 Broker 发送CommandPartitionedTopicMetadata请求，拿到这个 topic 配置的分区数。这一步是网络往返。所以生产者的创建默认是异步的，create()实际上是createAsync().join()的同步阻塞版本。

分区数决定生产者类型：如果 partitions > 1，创建的是ProducerMultiplexer（或者叫分区生产者），内部持有多个ProducerImpl，每个分区一个；如果 partitions = 0 或 1，直接创建单个ProducerImpl。这个设计意味着，分区的路由逻辑在生产者内部就完成了，对用户完全透明。

2.2 连接建立与 Producer 注册

拿到分区生产者后，核心逻辑在ProducerImpl（或ProducerMultiplexer）中。每个ProducerImpl需要和 Broker 建立连接并注册自己，这个动作在connectionOpened()方法里完成：

// ProducerImpl.connectionOpened() 核心流程 public void connectionOpened(ClientCnx cnx) { // 1. 先注册 Producer 到连接 cnx.registerProducer(producerName, producerId, this); // 2. 发送 CommandProducer 给 Broker cnx.sendCommandWithPayload( CommandProducer.builder() .topic(topic) .producerId(producerId) .producerName(producerName) .requestId(requestId) .encrypted(encryptionKey != null) .metadata(metadata) .build(), metadataPayload); }

这里有个值得注意的细节：CommandProducer命令里包含了encrypted标志和 metadata。如果启用了消息加密，Broker 在后续的消息收发中会走加密通道。但真正发送数据时，用的是另一个方法cnx.sendCommandWithPayload()，它会把消息的 payload 直接附在命令后面一起发送——这就是 Pulsar 协议的精髓之一：命令和载荷合并在同一个二进制帧里，减少网络往返次数。

注册成功后，Broker 返回CommandProducerSuccess，包含producerName（Broker 分配的生产者名称）。之后生产者状态从Connecting变为Ready，本地也生成了producerName和producerId，这两个 ID 在后续所有发送请求中都会带上。

2.3 状态机与关键字段

ProducerImpl内部维护一个状态集合，核心状态包括Ready、Connecting、Registering、Closed、Terminated。发送消息之前，代码会做一次状态检查：

private <T> void checkState() { if (state != State.Ready) { throw new PulsarClientException.NotConnectedException(); } }

这个检查必须在入队之前完成。如果生产者还在Connecting状态就发送，直接抛NotConnectedException。这个设计避免了一个经典问题：客户端消息发出去才发现没连上，导致消息在 TCP 缓冲区里丢失。先检查状态再入队，至少保证消息进入有序队列时连接是可用的。

提示：生产环境中我看到很多人误把create()的耗时当成生产者的"连接耗时"，其实不是。create()返回时 Producer 已经是 Ready 状态，因为它是同步等待元数据和注册完成。如果你用createAsync()，注册完成的回调在connectionOpened()里触发，不能提前发消息。

3. send() 的走向：分区选择、队列缓冲与批处理

3.1 分区选择器：消息去哪里，一开始就定了

分区选择逻辑由Partitioner接口定义，默认实现是DefaultPartitioner。send()最终会走到PartitionedProducerImpl（如果分区数>1）或者ProducerImpl（单分区）。

DefaultPartitioner的算法不复杂：

public int partition(String topic, byte[] key, byte[] data) { if (key == null) { // 无 key：轮询选择分区（round-robin） int currentPartition = Math.abs(topicHash % numPartitions); topicHash += hash(); return currentPartition; } else { // 有 key：一致性哈希 return Math.abs(Murmur3_32Hash.getInstance().makeHash(key) % numPartitions); } }

注意两个细节：

• 无 key 时用轮询，带 key 时用 key 的哈希。这保证了相同 key 的消息永远进同一个分区，从而保证同一 key 的消息在 partition 内严格有序。
• 哈希算法是可插拔的。Pulsar 提供Murmur3_32Hash和JavaStringHash两种内置实现，你可以实现自己的Hash接口来替换默认哈希。如果你的业务需要 URL 级别的一致性路由，这其实是个半隐藏的扩展点。

分区选择发生在内存缓冲之前，因为分区信息是消息入队的必要参数——不同的分区是不同队列，不能混。

3.2 内存缓冲：pendingMessages 队列

入队动作在ProducerImpl.internalSendAsync()里，这是整条链路的关键方法。它的核心是pendingMessages队列，一个ArrayDeque<OpSendMsg>：

private final ArrayDeque<OpSendMsg> pendingMessages = new ArrayDeque<>(); private <T> CompletableFuture<MessageId> internalSendAsync(Message<?> message) { // 构造 OpSendMsg OpSendMsg op = OpSendMsg.create(message, callback, partition, producerName); // 写入队列 pendingMessages.add(op); // 触发发送任务 cnx.getConnection().getEventLoop().execute(() -> { sendMessage(producerName, op); }); return op.getFuture(); }

这里有三件事值得停下来分析：

第一，消息以OpSendMsg对象入队。这个对象封装了MessageImpl、用户回调、发送时间、消息 ID。它是"一次发送操作"的完整描述。注意消息还没有被序列化，只是把 Java 对象放进队列，真正的序列化发生在后面批次构建时。

第二，队列是纯内存的。如果 Broker 响应慢，消息会在pendingMessages里积压。这个队列的长度受maxPendingMessages限制（默认 1000），超过就抛ProducerQueueIsFullError。这个限制非常重要，它是客户端唯一的背压机制——如果 Broker 写入速度跟不上生产速度，消息会在客户端排队，而不是无限发到网络。

第三，发送动作是异步的。send()返回CompletableFuture，消息是否成功取决于 Broker 是否返回SendReceipt。这一步是整个 Pulsar 客户端"纯异步"特性的基石：所有 IO 都在 Netty 的 event loop 线程上执行，用户线程只负责入队。

3.3 批处理：一条 Send 命令塞一堆消息

批处理是 Pulsar 生产者性能的关键。ProducerImpl里的sendMessage()方法最终走到sendAsyncWithBatch()，批处理逻辑集中在这里：

private void sendAsyncWithBatch(OpSendMsg op) { // 1. 检查是否满足发送条件 if (canAddToBatch(op)) { // 2. 尝试加入当前批次 batchMessageContainer.add(op); } else { // 3. 批次满了或条件不满足，先刷出当前批次，再开新批次 MessageImpl<?> msg = batchMessageContainer.newMessageImpl(); pendingSendTimer.cancel(); cnx.sendCommandWithPayload(msg, ...); batchMessageContainer.add(op); } }

canAddToBatch()判断批次容量：检查当前批次的消息数（maxNumMessagesInBatch）和字节数（maxBytesInBatch）。默认配置是 1000 条消息或 128KB。任何一个阈值达到，批次就"关闭"并发送。

关键点在batchMessageContainer.add(op)。这里不是简单地把OpSendMsg塞进去，而是构建一个聚合后的MessageImpl：

public void add(OpSendMsg op) { // 把单条消息的 metadata 追加到批次 metadata 数组 msgMetadataBuilder.addSingleMessageMetadata( SingleMessageMetadata.builder() .payloadSize(payloadSize) .build() ); // payload 直接拼接到批次 payload buffer payloadBuilder.writeBytes(op.getMessage().getData()); // 记录每个消息的回调 messages.add(op); consumerStats.updateNumMsgsSent(...); }

这段代码我看了很长时间才理解它的精妙：批处理不是把消息串行化成一长串字节，而是"横向拼接"。每条消息的 metadata（比如序列号、时间戳、payload 大小）都作为一个独立的SingleMessageMetadata追加到数组里，而 payload 则是简单地拼接进一个大 ByteBuf。Broker 收到后，可以根据 metadata 数组拿到每条消息的边界，再逐条解析。

这样做的直接效果是：一次网络 write 操作可以携带几百甚至上千条消息，IO 效率成倍提升。但代价是单条消息的发送延迟会略微上升——因为消息要等到批次满或定时器触发才真正发出。

批次定时器是SendTimeout（默认 1ms），这个定时器放在ProducerImpl里，每 1ms 检查一次是否有未满批次需要强制发送：

timeoutTask = executor.scheduleAtFixedRate(() -> { if (batchMessageContainer.getCurrentBatchBytes() > 0) { cnx.sendCommandWithPayload(batchMessageContainer.newMessageImpl(), ...); } }, sendTimeoutMillis, sendTimeoutMillis, TimeUnit.MILLISECONDS);

所以"批处理延迟"最大也不会超过 1ms + 网络往返。你看到生产者的sendTimeoutMillis参数，默认就是 1ms——这也是很多用户感知不到批处理存在的原因，延迟足够低。

经验：如果你需要极低的 P99 延迟，可以把batchingMaxMessages调整为 1，或者直接batchingEnabled(false)，这样每条消息单独一个 Send 命令。我实测过，禁用批处理后吞吐会掉到原来的约三分之一，但 P99 可以从 5ms 降到 2ms 左右。哪个更重要，取决于业务。

3.4 队列与批处理的边界条件

批次设计的三个边界条件，源码里写得很清楚：

• maxPendingMessages（默认 1000）：队列中未确认消息上限。这里我强调是"未确认"，不是"未发送"。队列里既有等待组批的消息，也有已发送但等待回执的消息。超出这个上限，send()直接失败。
• maxPendingMessagesAcrossPartitions（默认 50000）：所有分区的生产者在高水位（high water mark）之上的总数限制。生产环境多个分区共享同一客户端时，这个值是全局背压的上限。
• sendTimeoutMs（默认 30000）：发送超时。如果 30 秒内消息还没有收到 Receipt，客户端会认为发送失败，触发重试或抛异常。

这三个参数其实构成了一个三级背压体系：队列上限（单生产者）→ 全分区上限（客户端总体验）→ 发送超时（网络层兜底）。理解这一层，你就能解释大部分生产问题。

4. 底层 IO：Netty、ClientCnx 与协议编解码

4.1 消息怎么变成二进制：Pulsar 协议帧

前面的分析停在cnx.sendCommandWithPayload()，这就是进入网络层的地方。ClientCnx是客户端与 Broker 之间的一个 Netty Channel，它维护了两个关键 Buffer：

protected ByteBuf getCommand() { ... } // 命令 buffer protected ByteBuf getPayload() { ... } // 数据 buffer

sendCommandWithPayload做的事是：

public void sendCommandWithPayload(CommandSend cmd, ByteBuf payload) { ByteBuf command = cmd.toByteBuf(); // 组装协议头 command.writeInt(command.readableBytes() + payload.readableBytes()); // 总长度 command.writeInt(cmd.getCommandType()); // 命令类型 command.writeInt(cmd.getRequestId()); // 请求ID // 写入 channel channel.writeAndFlush(command, ...); }

关键点：Pulsar 协议帧的结构是[totalSize][commandType][requestId][commandData][Payload]。totalSize是整个帧的字节数，commandType是命令枚举值（比如SEND= 2），requestId是消息的唯一标识。这个帧结构是手写二进制，不是 Protobuf 直接编码的——Protobuf 负责的是CommandSend内部字段的编码，而帧外层是定长的二进制头。

这里有个性能优化细节：写 flush 不是每条消息 flush 一次，Netty 的channel.writeAndFlush()在底层会调用ChannelOutboundBuffer.addMessage()，真正 flush 到 socket 发生在下一次 event loop 的flush()回调。所以即使你逐条调用sendCommandWithPayload，底层也可能把多条消息合并到一个 TCP 段里发送。

4.2 批次消息在协议层的样子

如果你启用了批处理，CommandSend里会包含一个特殊字段numMessagesInBatch。Broker 通过这个值来区分"单条消息"和"一个批次"。批次消息到达 Broker 后的处理路径：

// Broker 端进入 handler 的简化逻辑 case Send: { CommandSend cmdSend = (CommandSend) cmd; // 从 payload 中解析 metadata 和内容 MessageMetadata metadata = msgMetadata; ByteBuf payload = cmd.getPayload(); // 如果 numMessagesInBatch > 1，批量解包 if (metadata.getNumMessagesInBatch() > 1) { // 循环解析每个 SingleMessageMetadata for (int i = 0; i < numMessagesInBatch; i++) { SingleMessageMetadata subMsgMetadata = SingleMessageMetadata.parseFrom(payload); int payloadSize = subMsgMetadata.getPayloadSize(); ByteBuf subPayload = payload.readSlice(payloadSize); // 逐条写入存储 } } }

这个解包过程我多说一句：Broker 单条记录地写入 BookKeeper，但注意写入是顺序的。一个批次内的消息会连续写入同一个 ledger，这样在日志存储中天然具备局部性，读取时也能顺序调度。这就是 Pulsar 批处理买的"两面包"：既减少了客户端 IO，又优化了存储层的顺序写。

4.3 连接建立与握手细节

所有ClientCnx在连接建立后都会进行一次协议的"能力协商"，主要看CommandConnect里是否带AuthData和ProtocolVersion。这个握手在源码里是ClientCnx.handleConnect()：

protected ByteBuf handleConnect(CommandConnect connect) { // 记录协议版本 this.remoteEndpointProtocolVersion = connect.getProtocolVersion(); // 认证：如果配置了 auth，返回 auth challenge if (authentication != null) { // 调用 authentication.getAuthData() 生成 auth data CommandConnected.Builder builder = CommandConnected.newBuilder() .serverProtocolVersion(ProtocolVersion.v12.getNumber()); return newCommandConnected(builder); } }

握手完成后，Producer 才会注册。如果 Broker 和客户端的协议版本不兼容，生产者注册会失败——通常表现为PartitionedTopicMetadata请求成功，但CommandProducerSuccess迟迟不来。我排过的一个典型问题就是：客户端版本比 Broker 新，导致ProtocolVersion字段超出 Broker 支持的枚举范围，Broker 直接拒连。

4.4 回执与回调：接收 SendReceipt 的路径

消息入了 Broker，Broker 处理完成后会返回CommandSendReceipt。这个返回值在ClientCnx.handleSendReceipt()里被接收：

protected void handleSendReceipt(CommandSendReceipt sendReceipt) { ProducerImpl<?> producer = producers.get(sendReceipt.getProducerId()); if (producer != null) { // 根据 requestId 恢复消息的发送请求 OpSendMsg op = producer.getPendingSendRequest(sendReceipt.getRequestId()); // 标记成功 op.setSendReceipt(sendReceipt); producer.ackReceived(op); // 触发用户回调 op.getSendFuture().complete(MessageId.convert(sendReceipt.getMessageId())); } }

这里的MessageId转换需要特别注意：CommandSendReceipt返回的 messageId 是(ledgerId, entryId, partitionIndex)三元组，Broker 会把 partition 信息编码进 messageId 里。所以同一个 topic 不同分区返回的 messageId 分区索引是不同的，消费者端游标定位依赖这个细节。

收到回执后，pendingMessages队列里对应的OpSendMsg会被移除：

public void ackReceived(OpSendMsg op) { pendingMessages.removeFirstOccurrence(op); // 释放内存 if (msgReferenceCount != null) { msgReferenceCount.release(); } }

这个移除动作很关键：只有收到 Brokder 回执，消息才会从内存队列里释放，队列才能腾出新位置。如果 Broker 一直不返回回执（比如磁盘故障、网络分区），队列会逐步被占满，最终触发ProducerQueueIsFullError。

5. 消息生命周期中的异常与高级细节

5.1 发送超时与重试机制

sendAsync有一个独立的超时检测路径，不是依赖 TCP 超时。ProducerImpl初始化时创建了一个executor（单线程定时调度），用它来扫描待确认消息：

private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(...); // 每 sendTimeoutMs 扫描一次 pendingMessages scheduledTask = executor.scheduleAtFixedRate(() -> { long timeout = System.nanoTime() - sendTimeoutMs * 1_000_000L; for (OpSendMsg op : pendingMessages) { if (op.getSendTime() < timeout) { // 超时了，标记失败并移除 OpSendMsg msg = pendingMessages.remove(op); msg.getSendFuture().completeExceptionally( new PulsarClientException.TimeoutException("Send timeout")); } } }, 0, sendTimeoutMs, TimeUnit.MILLISECONDS);

超时机制有个设计取舍：它不做发送重试，而是直接抛异常。这与 Kafka 客户端的语义不同——Kafka 默认会重试retries次直到提交确认，而 Pulsar 把重试策略完全交给应用层。如果你需要重发，要么手动 catch 超时后重新send()，要么用sendAsync()在回调里处理。

注意：sendTimeoutMs超时后消息仍然在 Broker 里可能已经写入成功，只是回执回来迟了。所以你看到的"超时"不代表"丢失"，重发才可能导致重复。生产环境需要权衡：是接受超时后重发（可能重复），还是保守的接受至少一次语义。

5.2 MessageImpl 的序列化细节与内存管理

MessageImpl的序列化在toByteBuf()方法里完成。核心代码：

public ByteBuf toByteBuf() { // 1. metadata 单独编码 ByteBuf metadata = serializeMetadata(); // 2. payload 直接用原字节,不拷贝 ByteBuf payload = this.pooledPayload.retain(); // 3. 组合成完整帧 ByteBuf buf = Unpooled.buffer(metadata.readableBytes() + payload.readableBytes()); buf.writeBytes(metadata); buf.writeBytes(payload); return buf; }

注意retain()这一步：payload 不是拷贝，是引用计数增加。如果你在send()之前修改了消息体，之后send()发出的内容还是你修改后的内容——因为retain()只是增加了计数，底层的字节数组还是同一个。所以发送中的消息不要修改，否则可能遇到并发问题。

5.3 消息key 与顺序保证

关于消息有序性，Pulsar 提供了两种保证：

• 单分区单生产者是有序的：同一分区下，生产者的消息按send()顺序写入。因为pendingMessages是 FIFO 队列，批次也严格按入队顺序聚合。
• 多生产者/多分区不保证全局有序：不同生产者并发写同一分区，Broker 按收到请求的先后顺序写入，但这与"发送顺序"可能不一致。

有个隐藏点：ProducerImpl.sendAsync()里的internalSendAsync并不会对消息做任何排序处理，只是入队。真正保证顺序的是pendingMessages队列的单线程消费——只有一个 event loop 线程在从队列里取消息发送。所以即使你并发调用 sendAsync，底层也是顺序发送。这个语义在batchingEnabled=true时依然成立，批次内部消息不会乱序。

5.4 内存与连接异常的处理

最后说一个我踩过的坑。如果 Broker 突然重启，ClientCnx会触发channelInactive，所有 pending 的OpSendMsg都会收到异常回调：

public void channelInactive(ChannelHandlerContext ctx) throws Exception { // 标记连接关闭 connectionClosed(); // 遍历所有 producer，触发异常 for (ProducerImpl<?> producer : producers.values()) { producer.connectionFailed(new PulsarClientException .LookupException("Connection closed")); } }

这里的connectionFailed会遍历pendingMessages，把每个OpSendMsg的回调都completeExceptionally。如果客户端配置了retryEnabled和reconnectEnabled，Producer 会自动重连并重新注册，但之前 pending 的消息不会自动重发——这是 Pulsar 客户端和某些消息中间件的一个区别：它是"至少一次"+"失败回调"语义，重发逻辑需要应用自己实现。

我建议在生产环境为sendAsync()注册一个whenComplete回调，做重试计数，而不是用同步send()+ catch 的方式——因为同步send()一旦 throw，整个调用链会中断，很难做精细的退避重试。

6. 参数调优与生产实践建议

6.1 关键参数对照表

我把真正影响发送性能的参数整理成一张表，这些都是你调优时需要动手的地方：

6.2 一个真实场景的调参思路

假设你要写入一个 128 字节的小消息，每秒需要发 10 万条。单条消息的 payload 很小，瓶颈在 IO 次数而不在字节数。这时你应该：

1. 开大 batch 容量：batchingMaxMessages(5000)，让单次 IO 发送更多消息
2. 微调发送延迟：batchingMaxPublishDelay(2, TimeUnit.MILLISECONDS)，2ms 攒一批，每批 5000 条，每秒 50 个批次，每个批次 100KB，maxPendingMessages放宽到 20000
3. 开启压缩：compressionType(CompressionType.LZ4)，128 字节的小消息压缩后大概能压到一半，内存占用和网络带宽都省

反过来，如果你的消息体很大（比如 1MB），批次几乎是"一条就满"，这时候批处理没有意义，重点要放在maxPendingMessages上，否则内存会爆。1MB × 1000 条 = 1GB 内存，这个账很好算。

6.3 连接与线程模型的理解

生产者使用的 Netty event loop 线程数由ioThreads控制（默认 1）。所有生产者的 IO 都在同一个 event loop 线程上执行，这意味着单个 event loop 的瓶颈会直接影响所有生产者的发送性能。如果你创建了多个 producer 面向高吞吐 topic，建议把ioThreads提高到 CPU 核数。

但注意，提高ioThreads不会提高单条消息的发送顺序性保证——顺序性基于pendingMessages的单线程消费，而不是ioThreads。多个 event loop 只影响不同 producer 之间的并行度。

6.4 常见问题速查表

7. 实操中我觉得最值得记住的三件事

第一件事，理解"异步"和"回调"的本质。Pulsar 客户端的异步不是"等一会儿就完事"，而是"消息进入队列、触发 IO、回执驱动释放"这样一条完整的事件链。你调send()完了就释放内存资源，是不对的——send()返回的 future 完成之前，消息对象还在队列里占用引用计数。如果你发送的消息很大且量大，记得msgReferenceCount的释放时机。

第二件事，批处理是 Pulsar 性能的支点，但批处理和背压是配套的。光调大maxPendingMessages而不管 Broker 的写入能力，只会让问题延后爆发，最后表现为sendTimeoutMs超时批量失败。我见过不少团队把超时时间一路从 30s 调到 120s，其实是掩盖而非解决 broker 磁盘缓慢的问题。

第三件事，源码读到这里，你已经具备了排查生产问题的重要能力。比如看到 distinct 的报错信息，你能判断它来自checkState()还是TimeoutException；看到批量发送的表现，你能判断是canAddToBatch()拒绝还是sendTimeout触发。这种"报错-定位-源码"的连锁反应，比任何监控面板都直接。

这套链路我反复读过很多遍，每次都有新的收获。如果你也在跟 Pulsar 源码，建议从ProducerImpl.internalSendAsync()开始断点追一遍，配合抓包看CommandSend的二进制结构，理解速度会快很多。