企业官网建设流程全解析

做了这么多年中间件，消息队列换了一茬又一茬，从 RabbitMQ 到 Kafka 再到 Pulsar，每次迁移都有新的体会。最近在一个内部数据同步项目里，需要把多个业务系统的变更事件统一收集后转发给下游数据分析服务，技术选型时对比了一圈，最终敲定了 Apache Pulsar。项目标题虽然只写着“使用 PulsarMQ 进行收发消息”，但真上手你就会发现，收发消息这四个字背后藏着一整套概念体系和无数个可以踩的坑。这篇博文就把我这次从零开始对接 Pulsar 的完整过程、核心配置和实战中遇到的问题都梳理出来，给同样刚接触 Pulsar 的朋友一份能直接照着操作的参考。

先说结论：Pulsar 的收发消息和 Kafka、RabbitMQ 在 API 层面并不复杂，甚至可以说更简单，但它的分层架构、订阅模型、确认机制和存储设计决定了你在使用方式上需要有一些全新的认知。如果你只是跑通 demo，十分钟就够了；如果想在生产环境稳定地收消息、发消息、不丢消息、不重复消费，那这篇文章值得你读完。

1. 为什么是 Pulsar：选型思路与架构认知

1.1 从传统消息队列到 Pulsar 的选型逻辑

在做技术选型时，最容易陷入的误区是拿 Kafka 的性能指标直接套到所有消息场景上。我这个项目的核心需求是：多个生产者持续写入事件数据，下游有多个不同的消费者组按各自逻辑消费，既要支持流式处理，也要支持一些简单的队列语义，同时希望运维成本可控。最初考虑过 Redis Stream、RabbitMQ、Kafka，最后 Pulsar 胜出主要看中了几点：

第一，Pulsar 的存储计算分离架构。Broker 节点只负责消息的路由和分发，实际数据存储在 Apache BookKeeper 里。这意味着扩容时不需要像 Kafka 那样做数据重平衡，broker 层面可以独立伸缩。对中小团队来说，这种架构让运维省心很多。

第二，多租户模型非常自然。Pulsar 用 tenant 和 namespace 做资源隔离，每个业务线一个 tenant，一个 topic 一个 namespace，权限管理、配额限制、备份策略都可以精确到 namespace 级别。我在项目里给数据同步、日志收集、业务通知各建了一套 namespace，互不干扰，后面做权限审计也方便。

第三，订阅模式丰富。Kafka 只有消费组的概念，本质上是 shared 模式，而 Pulsar 提供了 exclusive、shared、failover、key_shared 四种订阅类型，能覆盖从严格顺序消费到高并发负载均衡的各种场景。这对我们这种一个 topic 多套消费逻辑的系统来说非常关键。

第四，消息确认机制设计合理。Pulsar 的 ack 是逐条确认的，消费者处理完一条消息后再发送 acknowledgment，broker 才会更新游标。这种机制天然支持消息重投，不像 Kafka 靠 offset 提交来间接实现。对于下游偶尔处理失败、需要重试的业务场景，这种模型明显更直观。

1.2 Pulsar 核心概念速览：别把 topic 想简单了

很多人上来就直接用命令行生产消息、消费消息，跑通了就觉得会了。其实收发消息之前，先花半小时理解 Pulsar 的核心概念，后面会少踩很多坑。

Pulsar 的消息模型分三层：tenant（租户）是最高层级，类似于一个隔离的实例；namespace 是第二层，类似于逻辑分组，一个 namespace 下可以有多个 topic；topic 是第三层，是实际收发消息的通道。完整的 topic 名称格式是persistent://tenant/namespace/topic，注意那个persistent前缀，它表示这是一个持久化 topic。Pulsar 也支持非持久化 topic，但生产环境几乎不用，因为消息不落盘，broker 重启就丢了。

Topic 本身在物理上会被划分为多个分片，也就是分区数（partition）。分区是消息并发和顺序性的基础：同一个分区内的消息保证顺序，不同分区之间不保证。所以如果业务要求严格顺序，要么使用单分区 topic，要么用 key_shared 订阅模式配合消息 key 把同 key 的消息路由到同一分区。

消息本身由四部分组成：消息 ID、消息 key、消息元数据（properties）和消息体（payload）。消息 ID 是全局唯一的逻辑标识，用于消息确认和查询；消息 key 决定消息被路由到哪个分区，同时也是 key_shared 订阅模式下分发给消费者的依据。理解这几个概念，后面排查消息丢失、重复消息、顺序乱掉这些问题时就能有的放矢。

1.3 四种订阅模式怎么选

订阅模式是 Pulsar 收发消息时最需要花心思决策的点。我整理了一个表格，方便对照选择：

我当时在数据同步场景用的是 Key_Shared，因为同一个业务主键的变更事件必须按顺序处理，但不同主键之间可以并行。另一个日志分析场景用的是 Shared 模式，因为日志消息之间没有顺序要求，只需要快速消费。如果你拿不准，默认选 Shared 就好，它最通用、并发能力最强，只是需要接受消息处理顺序是不确定的这个前提。

2. 环境准备：本地快速搭建一套可用的 Pulsar

2.1 用 Docker 启动单机版 Pulsar

刚开始验证收发消息，没必要一上来就搭集群。Pulsar 官方提供了 standalone 模式，一个进程就能跑完整功能。如果你本机有 Docker，用官方的 apachepulsar/pulsar 镜像最省事：

docker run -d \ --name pulsar \ -p 6650:6650 \ -p 8080:8080 \ apachepulsar/pulsar:3.3.0 \ bin/pulsar standalone

这里有两个端口要记住：6650 是客户端收发消息的 TCP 端口，8080 是 Admin 接口的 HTTP 端口。客户端连接地址是pulsar://localhost:6650，管理工具使用http://localhost:8080。

如果你是那种不喜欢容器、更信任原生二进制的开发者，也可以去官网下载 release 包，解压后直接运行bin/pulsar standalone，效果是一样的。注意 standalone 模式下数据和元数据会存在本地磁盘，默认在data/目录下，想重置环境时删掉这个目录再重启即可。

启动完成后，Pulsar 默认会在public/default这个 namespace 下准备好两个系统 topic，分别是persistent://public/default/mynamespace之类的内部 topic，不用理会。自己测试时直接用persistent://public/default/my-topic这种名字就行，不需要提前创建 topic，Pulsar 支持自动创建。但生产环境我强烈建议关闭自动创建，避免手滑写错 topic 名导致数据进错通道。

2.2 用 pulsar-client 命令行验证收发消息

环境起来之后，先不要急着写代码，用 Pulsar 自带的命令行工具pulsar-client验证一下链路是否通。这步非常关键，它能帮你区分问题出在环境还是代码。

先打开一个终端，订阅并消费消息：

bin/pulsar-client consume \ "persistent://public/default/quickstart-topic" \ -s "test-subscription" \ -n 5

参数说明：-s指定订阅名称，-n指定消费多少条后退出。这个命令会阻塞等待消息，不要急着关。

再打开另一个终端，生产消息：

bin/pulsar-client produce \ "persistent://public/default/quickstart-topic" \ -n 10 \ -m "hello pulsar"

执行完生产命令后，回到消费终端，你应该能看到 10 条消息被逐个打印出来。如果这一步跑通了，说明 broker、bookkeeper、topic 都在正常工作，接下来写代码就会有信心得多。

这里注意一个细节：消费端第一次连接时，如果订阅不存在，Pulsar 会自动创建订阅。但这个订阅默认是持久化的，它会从最早的消息开始消费。这就是为什么我建议先用命令行验证，因为命令行整洁的输出会直接告诉你是谁在消费、消费了几条、消息内容是什么。用最小环境排除掉环境因素后，代码里的 bug 就更好定位了。

2.3 用 pulsar-admin 管理 topic 和订阅

除了收发消息，日常运维和验证环境时pulsar-admin也是高频工具。比如查看某个 topic 的统计信息：

bin/pulsar-admin topics stats "persistent://public/default/quickstart-topic"

这条命令会返回 topic 的生产速率、消费速率、积压消息数量、订阅列表等完整指标。我用它排查过消费者断连后消息积压的问题，一眼就能看出是哪个订阅卡住了。

还有几个实用的管理操作：

# 查看所有 topic bin/pulsar-admin topics list public/default # 重置某个订阅的消费位置到最早 bin/pulsar-admin topics reset-cursor \ "persistent://public/default/quickstart-topic" \ -s "test-subscription" # 卸载某个 broker 上的负载 bin/pulsar-admin brokers unload my-broker-1

reset-cursor 这个命令在测试阶段极其好用。我之前跑消费逻辑时改了几次业务处理代码，想重新消费一遍历史数据，直接 reset 游标就行，不用重新生产消息，大幅提高调试效率。

3. 代码对接：Producer 与 Consumer 的实操全解析

3.1 引入依赖与确定连接参数

我们团队的服务大部分是 Java 技术栈，这里就以 Java 客户端为例。Maven 引入依赖非常简单：

<dependency> <groupId>org.apache.pulsar</groupId> <artifactId>pulsar-client</artifactId> <version>3.3.0</version> </dependency>

版本号建议和你的服务端版本保持一致，避免客户端与服务端协议版本不兼容。我遇到过客户端版本比服务端新太多导致连接被拒的情况，排查了半天才发现是两个版本对协议协商的处理有差异。

然后创建客户端：

PulsarClient client = PulsarClient.builder() .serviceUrl("pulsar://localhost:6650") .connectionTimeout(10, TimeUnit.SECONDS) .operationTimeout(30, TimeUnit.SECONDS) .build();

team 里有人问我这两个超时参数怎么配置。connectionTimeout 是建立 TCP 连接的超时，通常 5-10 秒合适，太短容易在 broker 负载高时误报失败；operationTimeout 是消息发送、订阅等操作的整体超时，涉及创建 producer/consumer 和查找 topic 元数据，30 秒比较稳妥。生产环境建议再配上.enableTls(true)和相应的证书配置，本地测试可以先跳过。

3.2 Producer 参数怎么配，为什么要这么配

创建 producer 的代码看起来很简单：

Producer<String> producer = client.newProducer(Schema.STRING) .topic("persistent://public/default/quickstart-topic") .create();

但真正的细节藏在那些可选的 builder 参数里。我把常用的几个参数及其背后的逻辑整理如下：

Producer<String> producer = client.newProducer(Schema.STRING) .topic("persistent://public/default/quickstart-topic") .producerName("my-producer-001") .sendTimeout(30, TimeUnit.SECONDS) .batchingMaxMessages(1000) .batchingMaxPublishDelay(5, TimeUnit.MILLISECONDS) .blockIfQueueFull(true) .maxPendingMessages(10000) .enableBatching(true) .create();

producerName最好显式指定，不然 Pulsar 会生成一串随机名字，排错时很难定位是哪台机器发的。自定义一个有业务含义的名字，生产端出问题时就能在 broker 后台日志里快速检索到。

sendTimeout决定消息发送的超时时间，默认是 30 秒。需要注意，超时并不代表发送失败，而是客户端放弃了等待 broker 的确认。如果这里设置太短，在 broker 瞬时抖动时会出现大量消息被标记为超时，实际上 broker 可能已经写盘成功，这会导致后续重发时产生重复消息。我的经验是：只要不是对延迟极度敏感的场景，保持 30 秒默认值即可，不要随便调小。

batching相关参数是 Pulsar 提供高吞吐的关键。默认情况下，producer 会把多条消息攒在一起批量发送，batchingMaxMessages控制单批最大消息数，batchingMaxPublishDelay控制最长等待时间。这两个参数需要配合着调：如果消息量大，可以增大batchingMaxMessages到 5000 甚至 10000；如果对延迟敏感，就把batchingMaxPublishDelay调小到 1 毫秒甚至关闭批处理。我当时做数据同步，单条消息就几 KB，延迟容忍度高，就把批处理开大，吞吐量提升明显。

blockIfQueueFull和maxPendingMessages是背压机制的核心。当发送队列满时，默认行为是直接抛异常，这会导致生产者代码频繁出错。我建议设成true，让发送逻辑阻塞等待队列腾出空间。这看起来像是降低了吞吐，实际上是保护系统不因为发送过载而雪崩。下游处理不过来时，宁可让生产者阻塞，也不要疯狂重试把 broker 打垮。

3.3 Consumer 的核心机制：从订阅到确认

Consumer 端的代码同样不复杂：

Consumer<String> consumer = client.newConsumer(Schema.STRING) .topic("persistent://public/default/quickstart-topic") .subscriptionName("data-sync-sub") .subscriptionType(SubscriptionType.KeyShared) .ackTimeout(30, TimeUnit.SECONDS) .receiverQueueSize(1000) .subscribe();

这行代码背后有几个关键机制需要理解清楚。

subscriptionType的选择直接影响消息的路由和消费语义，前面已经讨论过，这里不再展开。值得注意的是，一个订阅名表示一类消费者逻辑组。同一个订阅名下可以挂多个消费者实例，它们共享同一个消费游标。如果你重启了服务但使用了不同订阅名，那它会从头开始消费这个 topic，这是新手最容易误解的地方。

ackTimeout是消费端最容易踩坑的参数。它表示一条消息被接收后，如果在这个时间内没有确认，broker 会自动把消息重新投递给其他消费者。这个机制的本意是处理消费者宕机导致消息丢失，但在实际业务处理时间波动较大时，它会引起大量重复消费。比如你处理一条消息平均要 10 秒，但某些极端消息要 40 秒，这时如果把ackTimeout设为 30 秒，就会在消息还在正常处理时被重复投递。我的建议是：现代客户端都支持消息级别或 consumer 级别的ackTimeout调整，如果业务处理时间长且不稳定，考虑用pause/resume控制消费节奏，或者在处理完成后手动 ack，不要过度依赖自动重投。

receiverQueueSize控制本地缓存的消息条数，默认是 1000。这个值设得越大，消费端预取的消息越多，吞吐越高，但带来的问题是：如果消费者在处理过程中崩溃，这些预取且未确认的消息会被重新投递，重复处理的概率会增大。同时，大队列会让内存占用升高。我在高吞吐场景设过 5000，配合批量确认逻辑，性能不错；但在对数据准确性要求极高的场景，我会调小到 100-300，牺牲一些吞吐换取更少的重复概率。

消费的逻辑循环如下：

while (true) { Message<String> message = consumer.receive(5, TimeUnit.SECONDS); if (message == null) { continue; } try { String value = message.getValue(); String key = message.getKey(); // 业务处理逻辑 processEvent(key, value); consumer.acknowledge(message.getMessageId()); } catch (Exception e) { // 这里千万不能吞掉异常 consumer.negativeAcknowledge(message.getMessageId()); log.error("process message failed", e); } }

收到消息后先处理，处理成功再 ack，处理失败要调用negativeAcknowledge让消息重新进入待投递队列。这里有个细节值得注意：negativeAcknowledge不是立即重投，客户端默认会等待 1 分钟后才让这条消息重新参与投递，这是为了避免那个消费者刚失败又把消息投回给同一个消费者，形成死循环。如果需要更快的重投间隔，可以在 consumer builder 里设置.negativeAckRedeliveryDelay(5, TimeUnit.SECONDS)。

另一个我自己常用的技巧是批量确认。如果业务逻辑确认多条消息都成功处理了，可以调用consumer.acknowledge(MessageId)遍历确认，或者使用consumer.acknowledgeAsync()异步确认。异步确认可以减少发送 ack 请求的等待时间，显著提高吞吐，尤其在消息量大、单条消息处理时间短的场景里收益明显。

3.4 消息顺序、去重与幂等设计

聊完了收发机制，必须说一个绕不开的话题：消息重复和顺序。Pulsar 本身保证的是 at-least-once 语义，也就是说消息不会丢，但可能重复投递。我的项目对数据准确性要求高，所以消费端必须做幂等处理。

幂等最简单的做法是利用消息 ID 去重。Pulsar 的消息 ID 由 ledgerId、entryId、partitionIndex 和 batchIndex 组成，理论上全局唯一。我维护了一个 Redis 集合，存储最近处理过的消息 ID，处理前先判断是否已存在。这个方案的缺点是内存消耗大，不适合超大数据量。

更常规的做法是在业务数据里带业务主键，消费端用数据库的唯一索引来天然去重。比如数据同步场景，我把每条消息的 value 里包含业务主键字段，落库时使用INSERT ... ON DUPLICATE KEY UPDATE语义，这样即使一条消息被重复投递，最终写入数据库的结果也是幂等的。

顺序性方面，Pulsar 只保证单个分区内的顺序。如果使用 Key_Shared 模式，同一个 key 的消息会路由到同一个消费者实例，天然有序。这里要注意一个坑：消费者实例数量变化时，key 的映射关系会重新分配，虽然 Pulsar 尽力保证同一个 key 在重分配前的消息先完成 ack，但极端情况下仍然可能出现顺序抖动。对顺序要求极其苛刻的场景，建议配合生产端的maxPendingMessages和消费端的receiverQueueSize做限制，减小重分配的影响窗口。

4. 常见问题与排查技巧实录

4.1 连接失败与认证问题

症状：客户端连接报TimeoutException或AuthenticationException。

排查步骤：先确认 broker 端口是否通，telnet localhost 6650看是否能连上。再确认 serviceUrl 没写错，注意是pulsar://而不是http://。如果是远程环境，检查防火墙和安全组有没有放开 6650 端口。我遇到过的经典错误是把 8080 当客户端端口填了进去，因为 8080 看起来像服务端口，但 Pulsar 客户端只走 6650。

如果是认证问题，检查 broker 侧配置authenticationEnabled=true时，客户端必须设置认证参数：

Authentication auth = AuthenticationFactory.token("your-jwt-token"); PulsarClient client = PulsarClient.builder() .serviceUrl("pulsar://localhost:6650") .authentication(auth) .build();

JWT 的生成要使用管理员账号在服务端操作，这里不展开。需要提醒的是，本地测试时如果没开启认证，不要配置认证参数，否则会报AuthenticationException，混淆排查方向。

4.2 消费者收不到消息

症状：生产者发送成功，消费者一直阻塞在receive()方法上。

这种情况第一件事就是检查订阅是否存在且连接到了同一个 topic。最容易犯的错误是：消费者使用的订阅名和之前测试时一致，而 cursor 已经消费到了最新位置，所以你发射的新消息还没到，消费者不会收到旧消息。如果确认是这种问题，使用pulsar-admin topics reset-cursor把游标重置到指定位置，或者不使用已存在的订阅名重新订阅，就能继续收到消息。

还有一种情况是 topic 有多个分区，消费者默认只订阅了其中一部分。Pulsar 的 consumer 订阅一个多分区 topic 时，单个 consumer 默认会接收所有分区的消息。如果你手动指定了分区订阅，比如consumer.subscribe()时指定MessageId.earliest之类的逻辑，可能只订阅了部分分区。检查一下消费者日志里的 subject 列表，确认它是否覆盖了所有分区。

4.3 消费速度慢、消息积压

症状：pulsar-admin topics stats中积压消息数量持续增长。

消息积压本质上是消费能力跟不上生产能力。先看消费者日志里有没有频繁的重试和negativeAcknowledge调用，有的话说明业务处理逻辑经常失败，这是主要矛盾。如果没有明显错误，那就是吞吐瓶颈。此时可以尝试以下几种手段：

• 增加消费者实例数量。Shared 模式下加实例可以水平扩展消费能力；Key_Shared 模式下要注意 key 的数量是否大于消费者数量。
• 调大receiverQueueSize。让每个消费者预取更多消息，减少网络往返。
• 检查业务处理链路里的外部依赖是否成为瓶颈，比如消费端写数据库时数据库连接池不够用。
• 使用批量接收：consumer.batchReceive()一次拿一批消息，批量处理后再统一 ack，可以显著减少 per-message 开销。

我自己排查过一次积压，最后发现是消费端调用的下游 API 超时时间设置太长，每条消息阻塞了 30 秒才失败重试。把超时调到 3 秒并增加重试退避后，积压很快就消化了。所以说，积压问题很多情况下不是 MQ 的锅，而是业务链路里的木桶短板。

4.4 消息重复与乱序的排查思路

症状：下游收到同一业务主键的多条消息，或者同 key 消息的处理顺序错乱。

重复消费最常见的原因是 ack 超时重投。当消费者处理时间超过ackTimeout，broker 会把这条消息重新投递，此时旧消费者的处理结果可能还没落库，就会导致并发处理同一条消息。排查时先看消费者日志里有没有同一消息 ID 被处理多次的记录，有的话就需要调大ackTimeout或者改为显式 ack 流程。

乱序问题首先确认订阅模式。如果是 Shared 模式，消息顺序得不到保证，必须接受现实或者换 Key_Shared。如果已经是 Key_Shared 但仍然乱序，检查生产者发送时是否设置了消息 key。没设置 key 时，Pulsar 会把 key 当空字符串处理，所有消息都路由到同一个分区，大概率顺序不会乱，但也意味着没有并行度。设置了 key 后，同 key 的消息会路由到同一分区，消费者实例变化时可能短暂乱序，一般通过降低receiverQueueSize缩短重分配窗口。

4.5 快速问题速查表

5. 生产接入前必须做的三件事

环境验证跑通只是第一步，真正上生产之前，有几个事情我建议你一定要先做好，不然上线后大概率要加班。

第一，关闭 topic 自动创建。在 broker 配置文件里设置allowAutoTopicCreation=false，所有 topic 通过 admin API 提前创建好，并且规定好命名规范。这样能避免业务代码里误写了 topic 名而创建出一堆垃圾 topic，后续查数据的时候都不知道数据去哪了。

第二，给消息设置合理的 retention 和 TTL。Pulsar 的消息默认是消费完成就会被清理，但如果你后续要做数据回溯或者补数，就需要配置保留策略：

bin/pulsar-admin namespaces set-retention \ --size -1 \ --time 7d \ public/default

这条命令表示公共 namespace 下的消息最多保留 7 天，大小不限。注意这个配置是 namespace 级别的，会影响下面所有 topic。我建议按业务需求分层设置：核心数据同步的消息保留 7-15 天，日志类数据保留 2-3 天即可。

第三，配置消息积压告警和消费者 lag 监控。Pulsar 的 metrics 可以通过 Prometheus 采集，重点关注pulsar_consumer_backlog_unacked_messages和pulsar_storage_backlog_size这两个指标。我们当时的策略是积压超过 10 万条就告警，超过 50 万条就拉群处理。没有监控的情况下，消息悄悄积压到爆炸是常有的事，特别是半夜某个下游服务重启导致消费断掉，早上起来一看积压几百万条。

第四，消费者进程要加优雅停机。Pulsar 客户端提供了consumer.close()方法，进程退出前要调用它，broker 会收到消费者离线通知，把未 ack 的消息重新投递。如果直接 kill -9，那就只能靠 ack 超时机制兜底，会造成短暂重复消费。我写了一个 shutdown hook 来处理这个问题，实测对重复消费的减少效果很明显。

6. 复盘与经验沉淀

这次 Pulsar 对接折腾下来，第一个体会是：消息队列的使用难点从来不在 API，而在对模型的理解。Pulsar 的订阅模型、ack 机制、存储架构与 Kafka 和 RabbitMQ 差异很大，一开始我就吃了订阅名复用导致重放全部历史的亏，也吃过 ack 超时设置不当导致消息疯狂重复的亏。这些坑在官方文档里都有写，但只有真正遇到时才印象深刻。

第二个体会是：先验证再写代码永远是正确的节奏。本地用 pulsar-client 命令行把链路打通、用 pulsar-admin 查看统计信息之后，再写 Java 代码对接就会非常顺畅。排查问题时也要习惯先看 broker 日志和 topic 统计，再猜业务代码的问题，顺序反了会浪费大量时间。

第三个体会给刚开始接触 Pulsar 的朋友：不要把精力花在背诵参数上，而是先想清楚你的业务对消息可靠性、顺序性、吞吐量的要求到底是什么。要求不一样，同一个 topic 的配置方案可能截然不同。我们内部后来沉淀了一套 topic 配置规范，把消息按业务类型分了等级，不同等级对应不同的订阅模式、保留策略和监控阈值，这比每个人凭借感觉配置要靠谱得多。

如果这个项目后续有扩展需求，我打算探索一下 Pulsar 的 Function 和 IO Connector，把一部分简单的消息处理逻辑直接下沉到 Pulsar 内部完成，减少业务服务之间的直连调用。那会是另一个有意思的话题，等实践完成后再来分享。