Pulsar 收发消息入门：搞懂存算分离、订阅模式与可靠性机制

从零上手 Pulsar 收发消息我建议你先搞懂这四件事做消息队列选型的时候身边不少团队在 Kafka、RabbitMQ 之间反复纠结我反而更推荐先去试试 Apache Pulsar。理由很简单它把存储和计算拆开了broker 只管读写调度数据落在 BookKeeper 里这样扩容、故障恢复都比传统 MQ 省心得多。而且它原生支持多租户、分区延迟、消息重放一套系统能把队列和流两种场景都覆盖了。这篇文章适合谁就是那种对 Pulsar 有点耳闻、想快速把“生产消息、消费消息”这条链路跑通的人。我会从部署环境开始讲清楚 Java 客户端如何完成最基本的收发再深入聊消费模型里最容易出坑的订阅模式最后把我实际项目中踩过的消息可靠性问题一并摊开。内容不会太长篇大论讲架构原理而是尽量做到“每一步都能照着复现每个参数都解释为什么这么设”。1. 为什么 Pulsar 值得单独学一套 API很多人在选 MQ 时会先问RabbitMQ 已经很成熟了Kafka 吞吐也很高为什么还要花时间学 Pulsar我的判断标准很简单——看团队的长期维护成本和业务是否有多场景混合诉求。1.1 存算分离带来的运维红利Pulsar 最核心的设计是 Broker 和 BookKeeper 分离。Broker 不保存数据它只是把消息路由给 BookKeeper 节点完成持久化。这意味着你扩容时不需要搬数据新增 broker 节点就能分担读写压力BookKeeper 节点故障时剩余节点会自动补副本不用像 Kafka 那样做复杂的分区重平衡。我当时从 Kafka 迁移过来的第一个感受是broker 重启时间从分钟级降到了秒级因为 broker 本身无状态重启只需要重新连接 BookKeeper 即可。这种差异在长时间运维中非常明显尤其是遇到核心节点不可用的时候。1.2 一套系统同时覆盖队列与流举个例子某内容平台既需要“给一万个用户推送通知”又需要“把点击日志实时灌入数仓”。以往可能会拆成两个系统通知用 RabbitMQ日志用 Kafka。但 Pulsar 的订阅模型天然同时支持这两种语义——Exclusive 订阅像队列、Shared 订阅像工作队列、Reader 接口则像 Kafka 的消费者组。一套集群两种需求运维资源直接减半。这也是我在项目里做技术选型时最看中的一点。如果业务还处于快速变化期很难提前确定明天的消息模型是“广播”还是“点对点”Pulsar 这种灵活切换的能力就显得尤为重要。2. 先把环境跑起来本地搭建一套可用的 PulsarPulsar 的后台组件确实比 RabbitMQ 多一些但用 Docker 拉起来也就一条命令的事。为了后面客户端能直接连接我建议按下面的方式部署单机模式。2.1 用 Docker Compose 部署单机集群# docker-compose.yml version: 3 services: pulsar: image: apachepulsar/pulsar:3.2.1 container_name: pulsar-single hostname: pulsar-single ports: - 8080:8080 # HTTP 服务端口 - 6650:6650 # 客户端通信端口 environment: - PULSAR_MEM -Xms512m -Xmx512m -XX:MaxDirectMemorySize1g command: bin/pulsar standalone这个配置里我预留了内存参数因为默认情况下 Pulsar 在开发机上可能会吃掉好几个 G 的堆外内存限制一下能避免本地开发时机器整体卡顿。注意standalone模式会把 BookKeeper 和 broker 放在一个进程里适合本地联调。真要部署多节点测试还是得用bin/pulsar initialize-cluster-metadata初始化元数据再分别启动 bookie 和 broker步骤会更繁琐但能暴露更多问题。跑起来之后用docker ps看一下服务状态观察日志里出现Server started at之类的字样基本就成功了。这里有一点容易忽略Pulsar 有两个端口8080 是管理 REST 接口6650 才是客户端收发消息用的。2.2 用命令行先验证链路是否通环境起来后我先用命令行手工收一条验证整体链路没毛病再上代码。# 进入容器 docker exec -it pulsar-single /bin/bash # 创建一条临时租户/命名空间 bin/pulsar-admin tenants create demo bin/pulsar-admin namespaces create demo/ns # 向某个 topic 生产消息 bin/pulsar-client produce demo/ns/test-topic --messages hello, pulsar # 消费消息 bin/pulsar-client consume demo/ns/test-topic -s demo-sub -n 1如果你能正常看到 “hello, pulsar” 这行消息说明集群和网络都没问题接下来就可以放心写客户端了。3. Java 客户端从 producer 到 consumer 的最小闭环Pulsar 官方提供 Java、Python、Go、C 等多语言客户端。我这里用 Java 演示因为 Java 在生产项目里的占比最高而且生态最全。你不需要引太多的依赖核心就是一个pulsar-client。3.1 引入依赖与初始化客户端dependency groupIdorg.apache.pulsar/groupId artifactIdpulsar-client/artifactId version3.2.1/version /dependency客户端本身是线程安全的一个应用里创建一个共享实例就够了不要每个操作都 new否则连接数会暴涨。PulsarClient client PulsarClient.builder() .serviceUrl(pulsar://localhost:6650) .ioThreads(8) .build();ioThreads控制的是 Netty 的 IO 线程数量默认是 CPU 核数。本地开发可以不用管线上如果发现 broker 连接数异常可以适当调大但也不要盲目给几十个线程容易造成上下文切换开销增大反而降低吞吐。3.2 生产者发送模式的选择发送消息最常用的有send()、sendAsync()和批量发送我实际用下来sendAsync()加回调是性能和可靠性平衡最好的方案。同步发送虽然代码简单但吞吐会受限于网络往返时间纯异步又容易忽略发送失败时的重试处理。ProducerString producer client.newProducer(Schema.STRING) .topic(demo/ns/test-topic) .create(); // 同步发送适合需要确认结果的场景 producer.newMessage() .key(order-123) .value(payment success) .send(); // 异步发送适合高吞吐场景但一定要处理异常 producer.newMessage() .value(another message) .sendAsync() .thenAccept(msgId - System.out.println(sent: msgId)) .exceptionally(ex - { System.err.println(send failed: ex.getMessage()); return null; });关于 message key我需要强调一下key 不是用来做业务标识的它是用来决定消息落在哪个分区的。同一个 key 的消息必定被路由到同一个分区从而保证局部有序。如果你不关心有序性可以不设置 keyPulsar 会采用轮询的方式分发。批量发送是提升吞吐最有效的手段之一设置enableBatching(true)并配合batchingMaxMessages和batchingMaxPublishDelayMicros可以让小消息在内存里攒一批再一起发。但要注意开启批量发送后单条消息在客户端里会有一个短暂的延迟窗口对延迟极度敏感的业务要慎重。3.3 消费者几种订阅模式的差异消费者是 Pulsar 里最有趣的环节因为同样一个 topic订阅模式不同消费行为完全不同。我最初以为“多个消费者订阅同一个 topic”天然就是负载均衡后来才发现只有Shared模式才是这样。consumer client.newConsumer(Schema.STRING) .topic(demo/ns/test-topic) .subscriptionName(demo-sub) .subscriptionType(SubscriptionType.Shared) .subscribe();这段代码很简洁但背后隐藏着几个关键知识点我放到了下一章集中说明。3.4 一个极易踩坑的点listener 不输出日志用 listener 风格消费时很多人会直接在received方法里写System.out.println结果发现日志不打印。原因很简单——Pulsar 的 listener 运行在 Netty 线程池上你写在里面的System.out会被吞掉。正确做法是把日志交给业务线程池或者用log.info配合对应的日志配置更稳妥的办法是先用consumer.receive()在独立的循环里测试排除线程问题再上 listener。while (true) { MessageString msg consumer.receive(); System.out.println(received: msg.getValue()); consumer.acknowledge(msg); }这个循环虽然简单却是排查各种消费端问题的最佳出发点。把 listener 换成循环接收日志正常了就说明问题是线程上下文导致的不是消息压根没发过来。4. 深入消费模型为什么你的多个消费者并没有同时消费这一章是本文的核心也是我在真实项目里被问得最多的地方。很多人把 Pulsar 当成 Kafka 用一个 group 里开好几个消费者实例以为消息会均匀分发结果发现只有一个实例在消费。其实大部分人踩的坑是用了默认的Exclusive订阅模式。4.1 四种订阅模式的行为对比订阅模式支持多个消费者消息分发方式适合场景Exclusive否仅一个单消费者独享全部消息严格控制顺序的日志处理Failover是主消费者活跃备消费者待命主备容灾不要求并行处理Shared是消息轮询分发每个消费者分一部分高并发业务、并行扩缩容Key_Shared是相同 key 的消息永远发到同一消费者同一用户/订单的有序处理我当初刚开始学的时候用 Exclusive 起了四个消费进程结果只有一个进程在消费其他三个完全没动静。当时第一反应是“是不是消息没发出来”排查了半天最后查文档才发现问题出在订阅模式上。换到 Shared 之后消费者数量一多消息随即被分散到各个进程吞吐成倍增长。但这里又引入了一个新问题——消息确认。在 Shared 模式下消息没有被 ack 之前会一直视为“未消费”状态。你如果 A 消费者收到了它但迟迟不确认这条消息在超时之后还会被重新投递给别的消费者可能导致重复处理。4.2 消息确认超时与重投策略Pulsar 的 ack 机制不是“收到即确认”而是“处理后确认”。因此如果消费者处理消息异常有几种做法立即 ack 再补发适合“确认收到即可”的业务重复消息由下游幂等兜底。正常 ack 前加 try/catch处理失败不 ack让 Pulsar 在超时后自动重投。配合 negative ack调用consumer.negativeAcknowledge(msg)直接要求重新投递不用等超时。这里还有一个非常容易踩的坑默认的ackTimeout是 0也就是不自动重投。有些同学把消息receive()出来之后处理失败没有 ack 也没有 nack结果消息就一直挂在未确认队列里消费位点也不推进看着就像“卡住了”。我在项目里一般会把ackTimeout调成 3 到 5 秒同时结合negativeAcknowledge做失败重试。这样既不会让失败消息长期阻塞队列也不会因为重投太频繁导致风暴。具体调多少要看消息处理耗时规则是“处理耗时的中位数乘三再乘二”。4.3 顺序性从 Kafka 带来的思维惯性要改改Kafka 里同一分区的消息是严格有序的消费者只要保证单线程消费一个分区就能维持顺序。但 Pulsar 的 Shared 模式下多个消费者同时处理消息顺序自然无法保证。如果你需要“同一订单的若干条消息按顺序处理”应该用 Key_Shared 模式并保证这些消息都有相同的 key。注意Key_Shared 模式下消费者的数量不能超过分区数。假如一个 topic 只有 3 个分区你起了 8 个消费者那就有一部分消费者永远收不到消息。我见过有人拿 Shared 模式消费订单超时类消息结果出现“支付成功先于订单创建被处理”的乌龙。后来改成 Key_Shared按订单 ID 做 key问题立刻解决。这是个很典型的“模式选错导致业务错误”的案例。5. 可靠性进阶重试、死信与积压监控很多人把 Pulsar 跑通“收发消息”就认为万事大吉直到线上出现消费失败、消息越积越多才开始重视可靠性机制。这些机制虽然不参与基础收发流程但在生产环境中几乎每个都是必选项。5.1 消费重试与死信队列的配合Pulsar 官方没有像 RabbitMQ 那样内置“重试队列”但它提供了negativeAcknowledgedeadLetterPolicy的组合可以实现类似的效果。ConsumerString consumer client.newConsumer(Schema.STRING) .topic(demo/ns/test-topic) .subscriptionName(demo-sub) .subscriptionType(SubscriptionType.Shared) .deadLetterPolicy(DeadLetterPolicy.builder() .maxRedeliverCount(3) .deadLetterTopic(demo/ns/dlq-topic) .build()) .subscribe();maxRedeliverCount表示一条消息最多被重投多少次超过之后会被丢到指定的死信 topic。我第一次用的时候漏了一个细节deadLetterPolicy只在Shared和Key_Shared模式下才生效。用 Exclusive 模式消费死信策略不报错也不生效排查起来相当隐蔽。处理死信消息时我的建议是从死信 topic 里捞出来做补偿操作比如记录失败原因、转人工处理或落库后重放。不要把它当成垃圾桶直接丢弃毕竟能进死信的大概率是真正需要关注的业务异常。5.2 消息积压的定位与处理思路消息积压是所有 MQ 都会遇到的问题Pulsar 里定位它比较方便因为它自带监控指标。你可以用pulsar-admin一条命令看某个订阅的积压情况bin/pulsar-admin topics stats demo/ns/test-topic输出会包含每个订阅的msgBacklog注意这个单位是“条数”不是字节数。结合生产速率和消费速率就能判断是下游处理太慢还是上游发得太猛。常见的积压原因和处理办法消费者单条处理耗时长吞吐上不去增加消费者数量前提是 SubscriptionType 是 Shared。消费者处理异常导致消息反复重投白白消耗带宽检查 ack 逻辑看看是不是漏了 finally 确认。批量发送导致消费端一次拉取过多消息内存被打满调小receiverQueueSize限制消费端本地缓冲。5.3 延迟消息的两种用法延迟消息的需求很多比如下单后 30 分钟未支付自动关闭或者定时发送通知。Pulsar 并没有像 RocketMQ 那样内置多个延迟级别而是靠deliverAt或者deliverAfter来实现类似效果。// 5 秒后投递 producer.newMessage() .value(delay 5s) .deliverAfter(5, TimeUnit.SECONDS) .send();这里我想补充一个重要认知延迟消息不是说消息本身在 broker 里“睡了一会儿”而是 broker 会把它放到一个不可见队列到时间后才变为可见消费者实际上感觉到的还是“一收到就能处理”。如果业务里大量使用这种延迟消息建议单独建一个 namespace 来区分避免普通消息被延迟消息拖慢。5.4 监控指标别只看 backlogPulsar 的指标里还有很多值得关注的比如publishLatency、ackRate、expiredMessageRate。我习惯在 Dashboard 里搭三块看板生产侧发布速率、发布失败率、消息大小分布。消费侧确认速率、消费积压、重投次数。存储侧BookKeeper 磁盘使用率、写入延迟。expiredMessageRate这个指标特别容易忽略。它表示“因为 ack 超时而重新投递的消息速率”。如果这个值一直很高说明有一批消息反复处理失败但你的业务日志里可能完全看不出问题因为这些消息可能卡在每个消费者的本地队列里。6. 实测过程中的几个典型问题与最终建议说完了理论我把实际测试过程中遇到的几个问题整理成表格方便你对照排查。这里面的问题都不是什么高深疑难杂症大多是“离实践就差一层窗户纸”的坑但每一条都曾让一个正常的服务在凌晨报警。现象根因解决方案起了 4 个消费者只有 1 个收到消息默认 Exclusive 订阅模式检查 subscriptionType改成 Shared消费端日志完全没输出listener 运行在 Netty 线程换成 receive 循环验证或委托业务线程池消息处理失败后一直不重投ackTimeout 默认 0显式设置 ackTimeout或用 negativeAcknowledge消费端明明收了很多消息但提交的位点没前进有消息没 ack统一在 finally 块里确认或用 cumulative acktopic 分区只有 3 个但起了 5 个消费者Key_Shared 下消费者不能超过分区数提前规划分区数或改用 Shared 模式延迟消息不准时比预期慢很多延迟队列在 broker 内单独调度积压会影响精度延迟消息单独 topic必要时提升 topic 优先级最后给几个我个人的实操建议。如果你还在学习阶段建议把 Pulsar 的客户端源码下载下来自己跑一遍例子尤其是samples目录里的 Producer、Consumer、Reader 示例。最开始不要追求用最新的 API稳定版本比新特性重要尤其是在生产环境。如果你已经开始在项目里集成我会建议“先跑通最小链路再慢慢加可靠性配置”。不要一上来就开批量、开死信、开延迟那样出问题时很难定位是哪个配置导致的。我自己的习惯是先能收到消息再加异步发送再加失败重试最后再接死信队列。每加一层我都会用模拟异常的方式验证一遍链路是否按预期走。我从 Kafka 切换到 Pulsar 之后有一段时间非常不适应它的订阅模型总觉得“一个 topic 不应该能同时被几种方式消费”。但真正用顺手之后我发现这种灵活性其实是双刃剑——好处是应对复杂业务游刃有余坏处是基础不扎实的人很容易用错模式还不自知。对于刚接触 Pulsar 的朋友我建议把重点放在“订阅模式的差异”和“消息确认机制”这两个知识点上这是收发消息之外的隐藏门槛。至于是用 Java 客户端、Go 客户端还是 Python 客户端其实影响不大API 设计基本一致理解模型才是一通百通的关键。