Apache Pulsar 从入门到精通

一、快速入门

Pulsar 是一个分布式发布-订阅消息平台，具有非常灵活的消息模型和直观的客户端 API。

最初由 Yahoo 开发，在 2016 年开源，并于2018年9月毕业成为 Apache 基金会的顶级项目。Pulsar 已经在 Yahoo 的生产环境使用了三年多，主要服务于Mail、Finance、Sports、 Flickr、 the Gemini Ads platform、 Sherpa (Yahoo 的 KV 存储)。号称：云原生消息队列之王

主要特征：

- 水平扩展

- 低延迟持久存储

- 多租户、认证、授权、配额

- 跨地域复制

- 主题的多种订阅模式（独占，共享和故障转移）。

官网：GitHub - apache/pulsar: Apache Pulsar - distributed pub-sub messaging system

B站入门视频： 01-Apache Pulsar 课程介绍_哔哩哔哩_bilibili

SpringBoot使用starter：GitHub - majusko/pulsar-java-spring-boot-starter: Simple pulsar spring boot starter with annotation based consumer/producer registration.

消息传递

Pulsar基于publish-subscribe(pub-sub)生产订阅模式，生产者将消息发布到Topic，消费者可以订阅这些主题来处理消息，并在处理完成后发送确认消息

消息生产者

发送模式

生产者将消息发布到Topic上，发送消息分可为同步发送和异步发送两种模式：

消息压缩
生产者发布消息在传输过程中会对数据进行压缩，目前Pulsar支持的压缩方式有LZ4，ZLIB, ZSTD, SNAPPY。如果启用了批处理，生产者将在单个请求中累积一批消息进行发送，批处理大小可以由最大消息数和最大发布延迟定义

批处理发送（Batching）
如果批处理开启，producer将会累积一批消息，然后通过一次请求发送出去。批处理的大小取决于最大的消息数量及最大的发布延迟。

消息消费者

消费模式

消费者从Topic上接收消息进行数据处理，同样，消息接收也分为同步接收和异步接收两种模式：

消费确认（ack）

1）消费者成功接收到消息时：

当消费者成功处理完一条消息后，会发送一个确认请求给broker，告诉broker可以删除这条消息了,否则broker会一直存储这条消息。消息可以逐个确认也可以累积确认，消费者只需要确认收到的最后一条消息，这个流中所涉及到的所有消息都不会再重新传递给这个消费者。

2）消费者不成功消费时

当消费者处理消息失败时，会给broker发送一个失败确认，这个时候broker就会给消费者重新发送这条消息，失败确认可以逐条发送，也可以累积发送，这取决于消费订阅模式。在exclusive和failover订阅模式中，消费者只会对收到的最后一条消息进行失败确认。在Pulsar客户端可以通过设置timeout的方式触发broker自动重新传递消息，如果在timeout范围内消费者都没有发送确认请求，那么broker就会自动重新发送这条消息给消费者。

3）确认超时

如果某条消息一直处理失败就会触发broker一直重发这条消息给消费者，使消费者无法处理其他消息，Dead letter topic机制可以让消费者在无法成功消费某些消息时接收新的消息进行消费，在这种机制下，无法消费的消息存储在一个单独的topic中（Dead letter topic），用户可以决定如何处理这个topic中的消息。

消息的持久化

消息的持久化是通过BookKeeper实现的，一旦创建了订阅关系，Pulsar将保留所有的消息（即使消费者断开了链接），只有当消费者确认已经成功处理保留的消息时，才会将这些消息丢弃消息。

消息的保留分为两种：

1.在保留策略内的消息即使消费者已发送了确认也可以持久地存储在Pulsar中，保留策略未涵盖的已确认消息将被删除，如果没有保留策略所有已确认的消息都将被删除；

2.设置消息到期时间，会根据应用于namespace的TTL过期时间，如果到期了，即使消息没有被确认也会被删除

当有某条消息被重复发送时，可以选择两种持久化策略：

1.是将重复的消息也持久化到BookKeeper中

2.是判断如果是重复消息，则不再进行持久化操作

租户（tenant）

Pulsar 从一开始就支持多租户，topic 的名称是层级化的，最上层是租户（tenant）

命名空间（namespace）

命名空间是租户内部逻辑上的命名术语。一个租户可以通过admin API创建多个命名空间。例如，一个对接多个应用的租户，可以为每个应用创建不同的namespace。

Topic

与其他pub-sub系统一样，Pulsar中的topic被命名为从生产者向消费者传输消息的通道：

{persistent|non-persistent}://tenant/namespace/topic

producer写入不存在的主题时了会在提供的命名空间下自动创建该主题。

常规topic只能由单个broker提供，这限制了topic的最大吞吐量，分区topic是由多个broker处理的一种特殊类型的topic，它允许更高的吞吐量。分区topic和普通topic在订阅模式的工作方式上没有区别，在创建主题时可以指定分区数。

消息路由模式

发布到分布分区topic主题时，必须指定路由模式。默认三个路由模式，默认轮询-和Kafka类似。

消息订阅模式（subscription）

Pulsar具有exclusive，shared，failover，Key_Shared 4种订阅模式

独占（exclusive）
exclusive模式：一个topic只允许一个消费者订阅，否则会报错

在 exclusive 模式下，一个 subscription 只允许被一个 consumer 用于订阅 topic ，如果多个 consumer 使用相同的 subscription 去订阅同一个 topic，则会发生错误。exclusive 是默认的订阅模式。如下图所示，Consumer A-0 和 Consumer A-1 都使用了相同的 subscription（相同的消费组），只有 Consumer A-0 被允许消费消息。

故障转移|灾备（failover）
failover模式：多个消费者订阅同一个topic，按照消费者名称进行排序，第一个消费者时唯一接收到消息的消费者（主消费者），当主消费者断开连接时，所有的后续消息都将发给下一个消费者

在 failover 模式下，多个 consumer 允许使用同一个 subscription 去订阅 topic。但是对于给定的 topic，broker 将选择⼀个 consumer 作为该 topic 的主 consumer ，其他 consumer 将被指定为故障转移 consumer 。当主 consumer 失去连接时，topic 将被重新分配给其中⼀个故障转移 consumer ，⽽新分配的 consumer 将成为新的主 consumer 。发⽣这种情况时，所有未确认的消息都将传递给新的主 consumer ，这个过程类似于 Kafka 中的 consumer 组重平衡（rebalance）。

如下图所示，Consumer B-0 是 topic 的主 consumer ，当 Consumer B-0 失去连接时，Consumer B-1 才能成为新的主 consumer 去消费 topic。

共享（shared）
shared模式：多个消费者订阅同一个topic，消息在消费者之间以循环的方式发送，并且给定的某条消息只能发送给一个消费者，当消费者断开连接时，所有发送给它但没有确认的消息将重新安排发送给其他消费者

在 shared 模式下，多个 consumer 可以使用同一个 subscription 去订阅 topic。消息以轮询的方式分发给 consumer ，并且每条消费仅发送给一个 consumer 。当有 consumer 失去连接时，所有发送给该 consumer 但未被确认的消息将被重新安排，以便发送给该 subscription 上剩余的 consumer 。

但是消息不能保证有序以及不支持批量ack

如下图所示，Consumer C-1，Consumer C-2，Consumer C-3 以轮询的方式接受消息。

共享键（key_shared）
key_shared模式：多个消费者订阅同一个topic，消息以分布方式在消费者之间传递（<key, value>），具有相同key的消息传递给同一个消费者，当这个消费者断开连接时，将导致key对应的消费者更改

在 shared 模式下，多个 consumer 可以使用同一个 subscription 去订阅 topic。消息按照 key 分发给 consumer ，含有相同 key 的消息只被发送给同一个 consumer 。

如下图所示，不同的 consumer 只接受到对应 key 的消息。

二、Pulsar原理架构

体系结构

在最高级别中，一个Pulsar实例有一个或多个Pulsar集群组成，实例中的集群可以彼此复制数据。在Pulsar集群中，一个或多个broker处理和加载来自生产者传入的消息，将消息发送给消费者，与Pulsar配置存储通信以处理各种协调任务，Pulsar集群架构如下所示，包括一个或多个broker，用于集群级配置和协调的Zookeeper，用于持久存储消息的BookKeeper，集群可以使用地理复制在集群间进行复制

Pulsar组件

Broker

Pulsar 的 broker 是一个无状态组件，本身不存储数据。主要负责处理 producer 和 consumer 的请求，消息的复制与分发，数据的计算。可以理解成Broker 是 Pulsar 的自身实例

主要有2部分组成：

1）HTTP服务器，向生产者和消费者公开，用于管理任务和topic查找端的REST API；

2）调度程序，异步TCP服务器，通过用于所有数据传输的自定义二进制协议；

每个集群都有自己的本地Zookeeper用于存储集群特定的配置和协调，如所有权元数据、代理加载报告、簿记员分类帐元数据等等。
Pulsar使用BookKeeper进行持久消息存储，BookKeeper是一个分布式预写日志（WAL）系统。

除了消息数据外，消费者的订阅位置cursor也可以持久地存储在BookKeeper中

每个 topic 的 partition 都会分配到某一个 borker 上，producer 和 consumer 则会连接到这个 broker，从而向该 topic 的 partition 发送和消费消息。broker 主要负责消息的复制与分发，数据的计算。

zookeeper

主要用于存储元数据、集群配置，任务的协调（例如哪个 broker 负责哪个 topic），服务的发现（例如 broker 发现 bookie 的地址）。

bookkeeper

主要用于数据的持久化存储。除了消息数据，cursors（游标）也会被持久化到 Bookeeper，cursors 是消费端订阅消费的位移。Bookeeper 中每一个存储节点叫做 bookie。

BookKeeper 是一种优化实时工作负载的存储服务，具有可扩展、高容错、低延迟的特点。企业级的实时存储平台应符合以下几项要求：

以极低的延迟（小于 5 毫秒）读写 entry 流
能够持久、一致、容错地存储数据
在写数据时，能够进行流式传输或追尾传输
有效地存储、访问历史数据与实时数据

数据存储

数据分区

写入主题的数据可能只有几个MB，也有可能是几个TB。所以，在某些情况下主题的吞吐量很低，有时候又很高，完全取决于消费者的数量。那么碰到有些主题吞吐量很高而有些又很低的情况该怎么处理？为了解决这个问题，Pulsar将一个主题的数据分布到多台机器上，也就是所谓的分区。

在处理海量数据时，为了保证高吞吐量，分区是一种很常见的手段。默认情况下，Pulsar的主题是不进行分区的，但通过命令行工具或API可以很容易地创建分区主题，并指定分区的数量。

在创建好分区主题之后，Pulsar可以自动对数据进行分区，不会影响到生产者和消费者。也就是说，一个应用程序向一个主题写入数据，对主题分区之后，不需要修改应用程序的代码。分区只是一个运维操作，应用程序不需要关心分区是如何进行的。【类似Kafka中的partition】

数据持久性

Pulsar broker在收到消息并进行确认之后，就必须确保消息在任何情况下都不会丢失。与其他消息系统不同的是，Pulsar使用Apache BookKeeper来保证持久性。BookKeeper提供了低延迟的持久化存储。Pulsar在收到消息之后，将消息发送给多个BookKeeper节点（具体由复制系数来定），节点将数据写入预写式日志（write ahead log），同时在内存里也保存一份。节点在对消息进行确认之前，强制将日志写入到持久化的存储上，因此即使出现电力故障，数据也不会丢失。因为Pulsar broker将数据发给了多个节点，所以只会在大多数节点（quorum）确认写入成功之后它才会将确认消息发给生产者。Pulsar就是通过这种方式来保证即使在出现了硬件故障、网络故障或其他故障的情况下仍然能够保证数据不丢失。

三、集群搭建

从官网下载bin包，修改配置，组件集群搭建模式即可。

Set up a standalone Pulsar locally | Apache Pulsar

四、客户端使用

1.申请消费组【内部限制】

2.申请token

Java使用相关

依赖Pulsar官方版本 2.9.5

<!-- in your <properties> block -->
<pulsar.version>2.9.5</pulsar.version><!-- in your <dependencies> block -->
<dependency><groupId>org.apache.pulsar</groupId><artifactId>pulsar-client</artifactId><version>${pulsar.version}</version>
</dependency>

完整的官方指南：

Pulsar Java client | Apache Pulsar

生产者配置

 public static void main(String[] args) throws PulsarClientException {// 1.1 建立客户端PulsarClientclient =PulsarClient.builder().serviceUrl("pulsar://abc.demo.com:6651") // 集群连接地址，请勿使用pulsar://host1,host2,host3的配置.authentication(AuthenticationFactory.token("************")).build();// 1.2 默认创建的生产者，schema为byte[]Producer<byte[]> producer = client.newProducer().topic("persistent://test/test/my-topic").compressionType(LZ4).enableBatching(true)  // 是否开启批量处理消息，默认true。需要注意的是enableBatching只在异步发送sendAsync生效，因此建议生产环境若想使用批处理，则需使用异步发送.batchingMaxMessages(10)    // 批处理中允许的最大消息数。默认1000，建议设成10.batchingMaxBytes(4*1024*1024) // 一批消息的最大大小。建议设成4MB，因为服务端有5MB的消息大小限制.batchingMaxPublishDelay(100, TimeUnit.MILLISECONDS) // 达到该时间就将一批消息发送出去，默认10ms，建议设大点100ms。可以理解为batch就算没攒够，一旦过了这么多时间还是会发送出去.maxPendingMessages(1000).create();// 1.3 同步发送单条消息MessageId messageId = producer.send("My message".getBytes());// 1.4 同步关闭生产者producer.close();client.close();System.out.println("发送pulsar消息成功:" + new String(messageId.toByteArray()));}

消费者配置

    public static void main(String[] args) throws PulsarClientException {// 1.1 建立客户端PulsarClientclient =PulsarClient.builder().serviceUrl("pulsar://abc.demo.com:6651") // 集群连接地址，请勿使用pulsar://host1,host2,host3的配置.authentication(AuthenticationFactory.token("************")).build();// 1.2 创建消费者Consumer<byte[]> consumer = client.newConsumer(Schema.BYTES)  // 消息schema，需要与生产者配置一致，默认是byte[].topic("persistent://test/test/my-topic").subscriptionName("my-subscription").subscriptionType(SubscriptionType.Shared)   // 指定消费模式，包含：Exclusive，Failover，Shared，Key_Shared。默认的是Exclusive模式.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)  // 指定创建新订阅时cursor的初始位置，默认Latest.subscribe();// 同步接收while (true) {Message msg = consumer.receive();try {System.out.println("Message received: " + new String(msg.getData()));consumer.acknowledge(msg);      // 消息需要ack了，才不会再次接收} catch (Exception e) {consumer.negativeAcknowledge(msg);}}}

注意：topic写上全路径名称，如 persistent://test/test/test/web-test2-topic；如果不填，就默认到这个ns下：persistent://public/default/my-topic

SpringBoot Starter使用

GitHub - majusko/pulsar-java-spring-boot-starter: Simple pulsar spring boot starter with annotation based consumer/producer registration.

1.引入依赖

<dependency><groupId>io.github.majusko</groupId><artifactId>pulsar-java-spring-boot-starter</artifactId><version>1.1.2</version></dependency>

2.定义生产者对象

@Configuration
public class TestProducerConfiguration {@Beanpublic ProducerFactory producerFactory() {return new ProducerFactory().addProducer("my-topic", MyMsg.class).addProducer("other-topic", String.class);}
}

在使用时可以通过模块来使用：

@Service
class MyProducer {@Autowiredprivate PulsarTemplate<MyMsg> producer;void sendHelloWorld() throws PulsarClientException {producer.send("my-topic", new MyMsg("Hello world!"));}
}

3.配置消费者

@Service
class MyConsumer {@PulsarConsumer(topic="my-topic", clazz=MyMsg.class)void consume(MyMsg msg) {// TODO process your messageSystem.out.println(msg.getData());}// 批量消费@PulsarConsumer(topic = "my-topic",clazz=MyMsg.class,consumerName = "my-consumer",subscriptionName = "my-subscription",batch = true)public void consumeString(Messages<MyMsg> msgs) {msgs.forEach((msg) -> {System.out.println(msg);});}// 批量消费和自动确认@PulsarConsumer(topic = "my-topic",clazz=MyMsg.class,consumerName = "my-consumer",subscriptionName = "my-subscription",batch = true)public List<MessageId> consumeString(Messages<MyMsg> msgs) {List<MessageId> ackList = new ArrayList<>();msgs.forEach((msg) -> {System.out.println(msg);ackList.add(msg.getMessageId());});return ackList;}// 批量消费和手动确认@PulsarConsumer(topic = "my-topic",clazz=MyMsg.class,consumerName = "my-consumer",subscriptionName = "my-subscription",batch = true,batchAckMode = BatchAckMode.MANUAL)public void consumeString(Messages<MyMsg> msgs,Consumer<MyMsg> consumer) {List<MessageId> ackList = new ArrayList<>();msgs.forEach((msg) -> {try {System.out.println(msg);ackList.add(msg.getMessageId());} catch (Exception ex) {System.err.println(ex.getMessage());consumer.negativeAcknowledge(msg);}});consumer.acknowledge(ackList);}// 消费使用到元数据@PulsarConsumer(topic="my-topic", clazz=MyMsg.class)void consume(PulsarMessage<MyMsg> myMsg) { producer.send(TOPIC, msg.getValue()); }// 覆盖默认消费名称@PulsarConsumer(topic = "my-topic",clazz = MyMsg.class,consumerName = "my-consumer",subscriptionName = "my-subscription")// 支持 spel 表达式@PulsarConsumer(topic = "${my.custom.topic.name}",clazz = MyMsg.class,consumerName = "${my.custom.consumer.name}",subscriptionName = "${my.custom.subscription.name}")public void consume(MyMsg myMsg) {}
}

4.配置必要参数

#PulsarClient
pulsar.service-url=pulsar://localhost:6650
pulsar.io-threads=10
pulsar.listener-threads=10
pulsar.enable-tcp-no-delay=false
pulsar.keep-alive-interval-sec=20
pulsar.connection-timeout-sec=10
pulsar.operation-timeout-sec=15
pulsar.starting-backoff-interval-ms=100
pulsar.max-backoff-interval-sec=10
pulsar.consumer-name-delimiter=
pulsar.namespace=default
pulsar.tenant=public
pulsar.auto-start=true
pulsar.allow-interceptor=false#Token based
pulsar.token-auth-value=43th4398gh340gf34gj349gh304ghryj34fh#Consumer
pulsar.consumer.default.dead-letter-policy-max-redeliver-count=-1
pulsar.consumer.default.ack-timeout-ms=3000

5.错误处理

@Service
public class PulsarErrorHandler {@Autowiredprivate ConsumerAggregator aggregator;@EventListener(ApplicationReadyEvent.class)public void pulsarErrorHandler() {aggregator.onError(failedMessage ->failedMessage.getException().printStackTrace());}
}

6.自定义拦截器

pulsar.allow-interceptor=true

Consumer Interceptor Example:

@Component
public class PulsarConsumerInterceptor extends DefaultConsumerInterceptor<Object> {@Overridepublic Message beforeConsume(Consumer<Object> consumer, Message message) {System.out.println("do something");return super.beforeConsume(consumer, message);}
}

Producer Interceptor Example:

@Component
public class PulsarProducerInterceptor extends DefaultProducerInterceptor {@Overridepublic Message beforeSend(Producer producer, Message message) {super.beforeSend(producer, message);System.out.println("do something");return message;}@Overridepublic void onSendAcknowledgement(Producer producer, Message message, MessageId msgId, Throwable exception) {super.onSendAcknowledgement(producer, message, msgId, exception);}
}

五、常见Q&A

1. 消费者消费不了消息，抛的异常和schema相关

检查消费者的schema和topic生产者的schema是否一致，如果不配置schema默认是byte[]字节流

Schema在new生产者或消费者的时候配置，例如下列demo

Producer<byte[]> producer = pulsarClient.newProducer()  // 默认byte[]
Consumer<byte[]> consumer = client.newConsumer(Schema.BYTES) Producer<User> producer = client.newProducer(JSONSchema.of(User.class))  // schema为json
Consumer<User> consumer = client.newConsumer(JSONSchema.of(User.class))