企业官网建设流程全解析

1. 先把名字说清楚：Pulsar 还是 pulsarMQ？

看到“pulsarMQ”这个写法，第一反应是又一个新出的消息中间件？实际不是。圈内人提到这个，通常指的就是 Apache Pulsar，一个开源的分布式消息流平台。Pulsar 并不是专门叫“pulsarMQ”的独立产品，而是把消息队列（MQ）和流处理能力合在了一起。你在项目里写pulsarMQ，多半是用了 Pulsar 的客户端，但又没完全把它的定位说清楚。

Pulsar 能做什么？简单说：生产者（producer）把消息发给主题（topic），消费者（consumer）从主题里拉消息，消息可以持久化、可以重放、可以多租户隔离。它既支持传统 MQ 的点对点队列，也支持类似 Kafka 的发布订阅流模式。比 Kafka 更友好的一点是，Pulsar 把存储层和计算层拆开了：broker 负责处理请求，bookie 负责存数据，两者可以独立扩容，不用像 Kafka 那样动不动就做数据重平衡。这也是我当初从 Kafka 转过来研究它的最大原因。

这篇文章适合谁看？适合准备上 Pulsar 做消息接入的开发者，或者已经在项目里了，但对收发消息、ACK 机制、积压排查还不够熟的工程师。我会从环境搭建开始，走一遍生产者、消费者的完整代码，再把可靠性和常见坑讲透，最后分享一些实际运维中才体会得到的经验。

2. 本地环境搭建：不用到集群部署，先把消息跑起来

2.1 两种启动方式：Docker Compose 和二进制包

我建议第一次接触 Pulsar 的人用 Docker Compose 起步，不需要关心 JDK 版本和系统依赖。一个简单的docker-compose.yml就能把 standalone 模式拉起来，只跑一个容器，里面同时包含 broker、bookie 和元数据服务。

version: "3.8" services: pulsar: image: apachepulsar/pulsar:latest container_name: pulsar-local ports: - "6650:6650" - "8080:8080" environment: - PULSAR_MEM="-Xms1g -Xmx1g" command: ["bin/pulsar", "standalone"]

启动命令：

docker compose up -d docker ps

Pulsar 的 broker 默认端口是 6650，客户端连接走这个端口；8080 是 HTTP 端口，用于管理 API 和运维页面。如果你想把数据持久化到本地，还要加一个卷：

volumes: - ./pulsar-data:/pulsar/data

如果你不想用 Docker，也可以下二进制包。官方发布包解压后执行：

tar -xzf apache-pulsar-*.tar.gz cd apache-pulsar-* bin/pulsar standalone

两种方式效果一样，但用二进制包时要注意环境变量JAVA_HOME必须正确配置，且默认 JVM 内存要求不低，开发机 1GB 堆内存起步。我在 2GB 内存的虚拟机上跑 standalone 会经常 OOM，后来把PULSAR_MEM调低才稳定。

2.2 验证安装：用命令行发一条消息

启动完成后，先别急着写代码，用自带的 CLI 工具验证一下环境。打开一个新终端，往里发一条消息：

bin/pulsar-client produce persistent://public/default/test-topic -m "hello pulsar"

在另一个终端里消费这条消息：

bin/pulsar-client consume persistent://public/default/test-topic -s my-subscription -n 1

参数-s指定订阅名，-n指定消费条数。成功的话，你会在消费者终端看到类似输出：

----- got message ----- hello pulsar

看到这里，整个链路就算通了。这一步很关键，因为后面调代码的时候如果出了连接问题，你可以先回退到命令行验证，快速定位到底是环境问题还是代码问题。

2.3 Topic 命名和默认租户：先说清 persistent 是什么意思

persistent://public/default/test-topic这个字符串里面包含四部分：persistent表示持久化主题，public是租户，default是命名空间，最后的test-topic才是主题名。Pulsar 里所有主题都归属到某个租户的某个命名空间下，这种设计在多人共用集群时非常好用——每个业务团队一个命名空间，数据隔离和权限控制都清晰。

对应的还有一种非持久化主题non-persistent://...。这种主题不落盘，只在内存中流转，消息重启即丢失。我一般只建议把非持久化主题用在实时监控、业务探活这类丢一两条也不影响的场景，核心业务数据一律走持久化主题。

3. 生产者实践：从同步发送到批量压缩

3.1 客户端依赖版本怎么选

Pulsar 的官方 Java 客户端包名是org.apache.pulsar:pulsar-client。版本上我建议新项目直接用 3.x 系列，它对应的协议版本较新，而且修掉了很多 2.x 的老问题。如果你维护的是老项目，至少也把版本升到 2.11 以上，不要再用那些很老的 API。

Maven 坐标：

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

如果是 Gradle：

implementation 'org.apache.pulsar:pulsar-client:3.3.1'

这里提醒一下：pulsar-client和pulsar-client-admin是分开的两个包。收发消息只需要前者，别误引了其他依赖导致 jar 冲突。

3.2 最基础的生产者代码

一个最简单的生产者长这样：

import org.apache.pulsar.client.api.*; public class PulsarQuickStartProducer { public static void main(String[] args) throws PulsarClientException { PulsarClient client = PulsarClient.builder() .serviceUrl("pulsar://localhost:6650") .build(); Producer<String> producer = client.newProducer(Schema.STRING) .topic("persistent://public/default/test-topic") .create(); producer.send("hello pulsar"); System.out.println("message sent"); producer.close(); client.close(); } }

ServiceUrl是 Pulsar 客户端连接 broker 的地址，本地用pulsar://localhost:6650。注意不是 HTTP 地址，除非你配的是 WebSocket 协议。

这里必须先说一个很容易踩坑的点：send()是同步方法，会一直阻塞到 broker 返回确认；如果你的业务是高性能写入，不要在高并发路径上直接调同步发送。我看到不少新手一上来就是for循环里挨个send()，整个系统的吞吐就被卡死了。

3.3 异步发送到底解决了什么问题

Pulsar 的客户端真正好用之处在于sendAsync()，它返回一个CompletableFuture，不会阻塞当前线程：

CompletableFuture<MessageId> future = producer.sendAsync("hello async"); future.whenComplete((messageId, ex) -> { if (ex != null) { System.err.println("发送失败: " + ex.getMessage()); } else { System.out.println("发送成功，消息ID: " + messageId); } });

异步发送的背后是有pendingQueueSize这个队列做缓冲的，默认 1000。如果生产速度超过 broker 处理速度，队列满了之后，sendAsync会主动 block 一段时间，不会无限膨胀内存。你要做的是设置合理的队列大小：

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

我实际测下来，同步发送的吞吐在单线程下也就几千 TPS，而异步发送配合多线程可以轻松到几万甚至更高。如果你的业务不是那种必须逐条等确认的场景，优先用异步。

3.4 批量发送：小消息的吞吐救星

每条消息都单独走一次网络往返，成本太高。Pulsar 支持在客户端本地攒批，攒到一定数量或间隔后再整体发出去。关键参数有三个：

• batchingMaxMessages：最大批量条数，默认 1000。
• batchingMaxBytes：最大批量字节数。
• batchingMaxPublishDelayMicros：最大攒批延迟，默认 1000 微秒。

配置示例：

Producer<String> producer = client.newProducer(Schema.STRING) .topic("persistent://public/default/test-topic") .enableBatching(true) .batchingMaxMessages(500) .batchingMaxPublishDelayMicros(10 * 1000) .compressionType(CompressionType.LZ4) .create();

我踩过的一个教训是：消息体积很小时，默认的攒批条件已经够用；但如果你的单条消息很大，比如几百 KB，就一定要调低batchingMaxBytes，否则一个批次会占用大量内存。批量发送能显著提升小消息的吞吐，但对大消息收益有限，这个要区分看待。

3.5 消息 key 和属性：不只是发出去就行

Pulsar 的消息对象可以带 key 和自定义属性。key 的作用是让相同 key 的消息落到同一分区，并保证在 Shared 订阅下也能维持同一 key 的顺序。如果业务上要求订单状态的变更严格按照订单号顺序处理，就把订单号设为 key。

TypedMessageBuilder<String> builder = producer.newMessage() .key("order-1001") .value("ORDER_PAID") .property("sourceSystem", "trade-service") .property("userId", "u-9527"); builder.send();

业务属性不要塞在消息体里用来做复杂解析，而是放在 property 里，消费者可以不解序列化完整消息就直接拿到这些字段。这个模式在日志采集、监控告警系统里非常常见。

4. 消费者实践：订阅模式、手动 ACK 和并发消费

4.1 三种订阅模式，先选对再写代码

Pulsar 的订阅模型从一开始就和 Kafka 不一样：一个主题上可以挂多个订阅，每个订阅独立管理自己的消费进度。这就像同一份报纸可以有好几个读者群，每个群的人都各看各的，互不干扰。

Pulsar 支持三种订阅模式：

我用一个生活化类比解释：Exclusive 是只有一把钥匙，谁拿到谁进门；Shared 是房间门口有好多把钥匙，谁去都能开一次门拿一条消息；Failover 是主钥匙坏了备钥匙自动顶上。

选择订阅模式关键是看你的业务要不要全局顺序。要严格顺序，只能选 Exclusive 或 Failover；要吞吐和水平扩展，选 Shared。需要注意，Shared 模式下不同消息可能由不同消费者处理，所以处理逻辑必须有幂等性，否则很容易出重复或错序问题。

4.2 消费者基本写法

最简单的 Shared 消费者：

import org.apache.pulsar.client.api.*; public class PulsarQuickStartConsumer { public static void main(String[] args) throws PulsarClientException { PulsarClient client = PulsarClient.builder() .serviceUrl("pulsar://localhost:6650") .build(); Consumer<String> consumer = client.newConsumer(Schema.STRING) .topic("persistent://public/default/test-topic") .subscriptionName("my-sub") .subscriptionType(SubscriptionType.Shared) .subscribe(); while (true) { Message<String> msg = consumer.receive(); try { System.out.println("received: " + msg.getValue()); // 业务处理... consumer.acknowledge(msg); } catch (Exception e) { consumer.negativeAcknowledge(msg); } } } }

这段代码里最关键的流程是：先receive()拿到消息，处理成功就acknowledge(msg)，处理失败就negativeAcknowledge(msg)，也就是负确认，让 Pulsar 稍后重新投递。很多新手会漏掉最后一步，结果消息一旦异常就直接丢失，或者一直没有 ACK 导致消费卡住。

4.3 手动 ACK、批量 ACK 与累计 ACK

Pulsar 支持三种确认方式：

• 单条确认：consumer.acknowledge(msg)，逐条确认。
• 累计确认：consumer.acknowledgeCumulative(msg)，把当前消息之前的所有消息一并确认。
• 批量确认：consumer.acknowledge(messages)，一次确认一批。

累计确认能减少网络请求数量，但有个前提：只适用于消息完全按顺序处理的场景。如果中间有一条消息处理失败，你把它跳过了，就不应该用累计确认，否则前面未处理的消息会被误认为已成功。

我自己的做法是：需要高吞吐、且每条消息和顺序无关时，用批量确认；严格有序、但处理失败要单独重试时，用单条确认加负确认。实际项目中，这两个场景我都遇到过后端数据库数据不对，排查了很久发现是 ACK 方式选错了。

4.4 并发消费：如何把吞吐抬起来

单线程receive()循环在大部分业务上都够用，但如果你想用满机器 CPU，就得搞并发。常见做法是使用共享消费模式，并启动多个消费线程：

ExecutorService executor = Executors.newFixedThreadPool(10); while (running) { Message<String> msg = consumer.receive(); executor.submit(() -> { try { handle(msg); consumer.acknowledge(msg); } catch (Exception e) { consumer.negativeAcknowledge(msg); } }); }

这种写法能把消费能力提升几倍，但要小心receiverQueueSize。它表示客户端本地缓存的消息数量，默认 1000。如果你开 10 个线程，实际上会把本地队列里的消息一把拉走，如果 broker 端积压很大，而本地消费速度跟不上，内存占用会飙升。

更稳妥的做法是每个消费线程持有独立的Consumer实例，让 broker 端参与负载均衡：

for (int i = 0; i < 4; i++) { Consumer<String> consumer = client.newConsumer(Schema.STRING) .topic("persistent://public/default/test-topic") .subscriptionName("my-sub") .subscriptionType(SubscriptionType.Shared) .subscribe(); executor.submit(() -> consumeLoop(consumer)); }

实测下来，四个消费者进程跑在同一个 JVM 里，比单消费者多线程的模式更稳定，重平衡逻辑也更清晰。

5. 可靠性机制：不丢、不重、不乱，Pulsar 是怎么设计的

5.1 消息存储在 BookKeeper，而不是 broker 内存

Pulsar 的持久化不是简单落个本地磁盘就完事。它把数据写到 Apache BookKeeper 的 bookie 节点上，写入时会指定副本数。你在 broker 上看到的 topic 实际对应一段段 managed ledger，消息以分段（segment）形式写入 bookie，客户端 ACK 后数据才真正落盘。

协议层里，生产者发送消息时可以设置级别，比如SendTimeout和MessageDeduplication，但核心可靠性靠的是 BookKeeper 的持久化保证。这也是 Pulsar 能在 broker 挂掉之后不丢数据的原因：broker 是无状态的，挂掉后其他 broker 接管客户端连接，而数据还在 bookie 里。

5.2 至少一次 vs 至多一次 vs 精确一次

Pulsar 默认提供的是“至少一次”语义：生产者发送消息后，如果不确定 broker 是否收到，可以重试，但重试可能造成重复消息。消费者处理时，如果处理成功但 ACK 消息丢了，消息也会重新投递。

这三者的关系可以这样理解：

5.3 幂等设计：这才是真正的“不重”

聊到精确一次，最重要的不是消息中间件，而是你的消费者业务逻辑。我见过太多团队纠结“能不能不重”，最后发现数据库没有唯一索引，消息重复进来就插了两条。

两个最实用的幂等方案：

• 消息里带唯一业务 ID，数据库表加唯一约束。
• 消费者用一个 Redis 缓存记录最近处理过的消息 ID，重复消息直接跳过。

Pulsar 消息自带MessageId，但它是物理位置的标识，不一定等于业务幂等 ID。所以我建议生产者在消息里放一个业务唯一键，比如orderId + eventType，消费端拿这个判断去重最靠谱。

5.4 消费进度和积压的管理逻辑

Pulsar 的消费进度存放在 broker 端的管理游标里，消费者每 ACK 一条消息，游标就往后移。未消费消息的数量叫 backlog。如果你发现 backlog 不断上涨，说明消费速度跟不上生产速度，或者消费者已经挂了。

Pulsar 有个好处是消息不会因为被消费就立刻删除，数据按保留策略进行清理。默认配置可能长期保留大量数据。这就意味着你可以随时通过一个新的订阅去重放历史消息，这在数据订正、故障排查时非常有用。Kafka 里重放历史靠重置 offset，Pulsar 里直接换个订阅名从头读即可。

6. 实操中遇到的常见问题与排查思路

6.1 消费者连接不了，报 Connection refused

最常见的原因是把地址写错了。本地 Docker 容器里启动 Pulsar，如果你用localhost:6650连不上，先确认容器端口映射是否正确：docker ps看 6650 端口是否暴露。另外，如果客户端也跑在 Docker 容器里，不能直接用localhost，要改成宿主机的局域网 IP 或者 Docker 网络内服务名。我之前把客户端和服务端都放 Docker Compose 里，写了localhost，结果死活连不上，改成pulsar:6650之后立刻通了。

6.2 消息发过去，消费者却收不到

先检查订阅名称是否一致。生产者只管往 topic 发消息，消费者需要先建订阅，如果消费者用的订阅名不同，消息不会推给当前消费者，而是积压在老订阅里。用 CLI 查看积压：

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

里面能看到msgBacklog字段。如果 backlog 在涨而消费者收不到消息，十有八九是订阅名配错了。

6.3 Shared 模式下消息顺序乱了

这不是 Bug，是 Shared 模式的固有行为。Shared 模式下，消息被平均分配给多个消费者，每个消费者处理速度不同，必然会导致顺序错乱。如果你要求严格顺序，要么换 Exclusive 或 Failover，要么把需要保证顺序的消息分区加上同一个 key，但这只能保证同一 key 的顺序，跨 key 之间依然没有全局顺序。

6.4 消费很久都没有 ACK，导致消息重复投递

Pulsar 默认有一个ackTimeout，如果消费者收到消息后在超时时间内没有 ACK，broker 会认为消息处理失败，重新投递给同订阅里的消费者。这条机制本来是为了防止消费者挂掉导致消息卡住，但也容易误伤慢任务。如果你有一些流程本身耗时超过默认 30 秒，记得把 ackTimeout 调大，或者干脆设置为 0 禁用超时重投。

.consumerName("slow-consumer") .ackTimeout(60, TimeUnit.SECONDS)

这里还要注意：ackTimeout不会自动让消费者跳过消息，而是重投。如果你既不想无限重投，又想跳过坏消息，可以结合死信主题使用：

.deadLetterPolicy(DeadLetterPolicy.builder() .maxRedeliverCount(3) .deadLetterTopic("persistent://public/default/test-topic-dlq") .build())

这样重试 3 次后，坏消息会进死信队列，主队列不受影响，后续可以手动排查死信里的问题消息。

6.5 生产端遇到“无法在队列中容纳消息”

报错信息类似于producer send queue is full，说明异步队列打满了。我遇到这种情况基本都是业务方不限制生产速率，或者 broker 写入能力不够。排查步骤：

1. 降低生产者并发。
2. 调大pendingQueueSize。
3. 检查 bookie 磁盘 IO 是否成为瓶颈。

记住，调大队列只是缓解症状，真正要查的是下游为什么消费跟不上。

6.6 使用的不是纯 Pulsar，而是比 Kafka 兼容 API

有些项目为了减少改造成本，用的是 Pulsar 的 Kafka 兼容 API，代码里写的是import org.apache.kafka.*。这样确实能少改代码，但有坑：Pulsar 的 Kafka 兼容协议和原版 Kafka 协议在一些行为细节上存在差异，比如事务支持、admin接口的覆盖范围。如果你已经在用 Kafka API，建议先跑一轮兼容性测试再上生产。

7. 最后分享几个我的实战体会

我自己从 Kafka 切到 Pulsar 的一年多里，最大的体验变化就是“不再害怕 broker 宕机”。Kafka 的 broker 节点带状态，挂了之后分区副本切换、数据平衡往往折腾半天；Pulsar 的 broker 是无状态的，换个节点重新连上就能继续消费，数据还在 BookKeeper 里躺着，这种感觉踏实很多。

如果要给刚接触 Pulsar 的人一个建议：先别急着搭集群。下载一个 schema 不复杂的 demo，本地把生产者、消费者跑通，然后把服务停掉再重启，体会一下消息不会丢的爽感。之后再把分区、订阅模式、ack 超时这几个概念一个个试一遍，你就能真正理解它的设计逻辑了。

还有一个必须提醒的细节：生产环境尽量用带persistent://前缀的主题，不要图省事用non-persistent://。因为非持久化主题的数据只存在内存里，broker 一重启，所有未消费的消息全部消失。很多团队在测试环境用惯了非持久化主题，迁到生产没注意，流量一上来丢了一堆重要消息，这个锅最后还得自己背。

再补充一个调试小技巧：用pulsar-admin命令检查主题的状态和积压，比看日志直观得多。常用的几条：

bin/pulsar-admin topics list public/default bin/pulsar-admin topics stats persistent://public/default/test-topic bin/pulsar-admin topics peek-messages -m 5 -s my-sub persistent://public/default/test-topic

peek-messages可以看到指定订阅里还没消费的消息内容，排查线上数据对不对的时候非常方便。我靠着这一条命令定位过不少问题，不用去翻日志捞 payload。

Pulsar 收发的路数就是这样，看起来简单，但细节很密。你踩过几次坑之后再回头看，会觉得这套设计其实很顺手，尤其是它把 MQ 的可靠性和流系统的高吞吐融合得相当自然。