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三种订阅模式

独占（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，集群可以使用地理复制在集群间进行复制