云原生消息队列 Pulsar 浅析——实践类

一、前言

Pulsar 是一个多租户，高性能的服务间消息解决方案。最初由 Yahoo 开发，现在由 Apache Software Foundation 负责。

Pulsar 是消息队列领域的一匹黑马，其最大优点在于它提供了比 Apache Kafka 更简单明了、更健壮的一系列操作功能，支持地域复制和多租户。此外，相比传统的 Kafka、RocketMQ 等，Pulsar 更加适合 IoT 的场景。

二、架构设计

2.1 整体架构

Apache Pulsar 和其他消息系统最根本的不同是采用分层架构。

Apache Pulsar 集群由两层组成：

无状态服务层：由一组接收和传递消息的 Broker 组成

有状态持久层：由一组 Apache BookKeeper 存储节点组成，可持久化地存储消息。

图片

Pulsar 客户端不直接与存储层 Apache BookKeeper 交互。客户端也没有直接的 BookKeeper 访问权限。这种隔离，为 Pulsar 实现安全的多租户统一身份验证模型提供了基础。

2.2 Broker

Broker 集群在 Apache Pulsar 中形成无状态服务层。Broker 不在本地存储任何消息数据。Pulsar 主题的消息，都被存储在分布式日志存储系统（Apache BookKeeper）中。

每个主题分区（Topic Partition）由 Pulsar 分配给某个 Broker，该 Broker 称为该主题分区的所有者。Pulsar 生产者和消费者连接到主题分区的所有者 Broker，以向所有者代理发送消息并消费消息。

如果一个 Broker 发生故障，Pulsar 会自动将其拥有的主题分区移动到群集中剩余的某一个可用 Broker 中。重点说明：由于 Broker 是无状态的，因此当发生 Topic 的迁移时，Pulsar 只是将所有权从一个 Broker 转移到另一个 Broker，并不会有任何数据复制发生，故障转移非常轻量。

下图显示了一个拥有 4 个 Broker 的 Pulsar 集群，其中 4 个主题分区分布在 4 个 Broker 中。每个 Broker 拥有并为一个主题分区提供消息服务。

图片

2.3 Bookeeper​

Apache BookKeeper 是 Apache Pulsar 的持久化存储层。Apache Pulsar 中的每个主题分区本质上都是存储在 Apache BookKeeper 中的分布式日志。

每个分布式日志又被分为 Segment 分段。每个 Segment 分段作为 Apache BookKeeper 中的一个 Ledger，均匀分布并存储在 BookKeeper 群集中的多个 Bookie 中。

通过 Segment 分段的方式，主题分区中的消息可以均匀和平衡地分布在群集中的所有 Bookie 中。这意味着主题分区的大小不仅受一个节点容量的限制；相反，它可以扩展到整个 BookKeeper 集群的总容量。

下面的图说明了一个分为 x 个 Segment 段的主题分区。每个 Segment 段存储 3 个副本。所有 Segment 都分布并存储在 4 个 Bookie 中。

图片

2.4 优势

计算存储分离的特性给 Pulsar 带来了许多特性

无限制的主题分区存储

无缝 Broker 故障恢复

无缝 Bookeeper 故障恢复

无缝集群扩展

无限制的主题分区存储

由于主题分区被分割成 Segment 并在 Apache BookKeeper 中以分布式方式存储，因此主题分区的容量不受任何单一节点容量的限制。

主题分区可以扩展到整个 BookKeeper 集群的总容量，只需添加 Bookie 节点即可扩展集群容量。

这是 Apache Pulsar 支持存储无限大小的流数据，并能够以高效，分布式方式处理数据的关键。

无缝 Broker 故障恢复

下图说明了 Broker 故障恢复。本例中 Broker2 因某种原因（例如停电）而断开。Pulsar 检测到 Broker2 已关闭，并立即将 Topic1-Part2 的所有权从 Broker2 转移到 Broker3。在 Pulsar 中数据存储和数据服务分离，所以当代理 3 接管 Topic1-Part2 的所有权时，它不需要复制 Partiton 的数据。如果有新数据到来，它立即附加并存储为 Topic1-Part2 中的 Segment x + 1。Segment x + 1 被分发并存储在 Bookie1, 2 和 4 上。因为它不需要重新复制数据，所以所有权转移立即发生而不会牺牲主题分区的可用性。

图片

无缝 Bookeeper 故障恢复

下图说明了 Bookeeper 的故障恢复。这里有一个磁盘故障导致存储在 bookie2 上的 Segment 4 被破坏。Apache BookKeeper 后台会检测到这个错误并进行复制修复。

图片

BookKeeper 中的副本修复是 Segment 级别的多对多快速修复，这比重新复制整个主题分区要精细，只要复制必须的数据。这意味着 Apache BookKeeper 可以从 bookie3 和 bookie4 读取 Segment4 中的消息，并在 bookie1 处修复 Segment4。所有的副本修复都在后台进行，对 Broker 和应用透明。

即使有 Bookie 节点出错的情况发生时，通过添加新的可用的 Bookie 来替换失败的 Bookie，所有 Broker 都可以继续接受写入，而不会牺牲主题分区的可用性。

无缝集群扩展

下图说明了 Pulsar 集群扩展。当 Broker2 将消息写入 Topic1-Part2 的 Segment X 时，将 Bookie X 和 Bookie Y 添加到集群中。Broker2 立即发现新加入的 Bookies X 和 Y。然后 Broker 将尝试将 Segment X + 1 和 X + 2 的消息存储到新添加的 Bookie 中。新增加的 Bookie 立刻被使用起来，流量立即增加，而不会重新复制任何数据。除了机架感知和区域感知策略之外，BookKeeper 还提供资源感知的放置策略，以确保流量在群集中的所有存储节点之间保持平衡。

图片

三、消息模型

3.1 通用的消息模型

消息模型一般以下 3 个方面：

• 消息消费：如何发送和消费消息

• 消息确认（ACK）：如何确认消息

• 消息保存（Retention）：消息保留时间，触发消息删除的原因以及怎样删除

3.2 通用的消费模型

队列模型

队列模型主要是采用无序或者共享的方式来消费消息。

通过队列模型，多个消费者可以从单个管道中接收消息；当一条消息从队列发送出来后，多个消费者中的只有一个（任何一个都有可能）接收和消费这条消息。消息系统的具体实现决定了最终哪个消费者实际接收到消息。

队列模型通常与无状态应用程序一起结合使用。无状态应用程序不关心排序，但它们确实需要能够确认（ack）或删除单条消息，以及尽可能地扩展消费并行性的能力。典型的基于队列模型的消息系统包括 RabbitMQ 和 RocketMQ。

流模型

流模型要求消息的消费严格排序或独占消息消费。

对于一个管道，使用流式模型，始终只会有一个消费者使用和消费消息。消费者按照消息写入管道的确切顺序接收从管道发送的消息。

流模型通常与有状态应用程序相关联。有状态的应用程序更加关注消息的顺序及其状态。消息的消费顺序决定了有状态应用程序的状态。消息的顺序将影响应用程序处理逻辑的正确性。

3.3 Pulsar 消息消费 

Pulsar 抽象出了统一的消费模型: producer-topic-subscription-consumer

Pulsar 的消息模型既支持队列模型，也支持流模型

Topic 是用于发送消息的通道。Topic 中的每条消息，可以根据消费者的订阅需求，多次被使用，每个订阅对应一个消费者组（Consumer Group）。每个 Topic 可以有不同的消费组。

消费者（consumer）被组合在一起以消费消息，每个消费组是一个订阅（subscription）

消费者可以拥有不同的消费方式：独占（Exclusive），故障切换（Failover）或共享（Share）

Pulsar 通过这种模型，将队列模型和流模型这两种模型结合在了一起，提供了统一的 API 接口。这种模型，既不会影响消息系统的性能，也不会带来额外的开销，同时还为用户提供了更多灵活性，方便用户根据自己的实际场景来使用消息系统。

独占订阅（流模型）

独占模式，topic 只能被一个消费者订阅。如果多于一个消费者以同样方式去订阅主题，消费者将会收到错误。

下图中，只有 Consumer A-0 可以消费

图片

容灾订阅（流模型）

容灾模式，多个消费者可以订阅同一个 topic，消费者按消息者名称的字典序排列。第一个消费者被初始化为唯一接收消息的消费者。这个消费者被称为主消费者（master consumer）。

当主消费者断开时，所有的消息（未被确认和后续进入的）将会被分发给下一个消费者。

在下图中，Consumer-B-0 是主消费者，如果 Consumer-B-0 断开连接，Consumer-B-1 会变成主消费者去接收消息

图片

共享订阅（队列模型）

在共享（shared）或轮询(round robin)模式下，多个使用者可以订阅同一个 topic。消息通过轮询方式分发给不同的消费者，并且每个消息仅会被分发给一个消费者。当消费者断开连接，所有被发送给它，但没有被确认的消息将被重新安排，分发给其它存活的消费者。

在下图中，Consumer-C-1 和 Consumer-C-2 可以订阅该主题，但是 Consumer-C-3 和其他消费者也可以订阅该主题。

图片

3.4 Pulsar 消息确认

当使用分布式消息系统时，可能会发生故障。比如在消费者从消息系统中的主题消费消息的过程中，消费者和 Broker 都可能发生错误。消息确认（ACK）的目的就是保证当发生这样的故障后，消费者能够从上一次停止的地方恢复消费，保证既不会丢失消息，也不会重复处理已经 ACK 的消息。

在 Pulsar 中，每个订阅中都使用一个专门的数据结构——游标（Cursor）来跟踪订阅中的每条消息的 ACK 状态。每当消费者确认消息时，游标都会更新。更新游标可确保消费者不会再次收到消息。

Pulsar 提供两种消息确认方法，单条确认（Individual Ack）和累积确认（Cumulative Ack）。通过累积确认，消费者只需要确认它收到的最后一条消息。主题分区中的所有消息（包括）提供消息 ID 将被标记为已确认，并且不会再次传递给消费者。

Pulsar 可以支持消息的单条确认，也就是选择性确认。消费者可以单独确认一条消息。被确认后的消息将不会被重新传递。下图说明了单条确认和累积确认的差异（灰色框中的消息被确认并且不会被重新传递）。在图的上半部分，它显示了累计确认的一个例子，M12 之前的消息被标记为 acked。在图的下半部分，它显示了单独进行 acking 的示例。仅确认消息 M7 和 M12 - 在消费者失败的情况下，除了 M7 和 M12 之外，其他所有消息将被重新传送。

图片

独占订阅或容灾订阅的消费者能够对消息进行单条确认和累积确认；共享订阅的消费者只允许对消息进行单条确认。单条确认消息的能力为处理消费者故障提供了更好的体验。对于某些应用来说，处理一条消息可能需要很长时间或者非常昂贵，防止重新传送已经确认的消息非常重要。

游标（Cursor）由 Broker 来管理，利用 BookKeeper 的 Ledger 提供存储。

Apache Pulsar 提供了灵活的消息消费订阅类型和消息确认方法，通过简单的统一的 API，就可以支持各种消息和流的使用场景。

3.5 Pulsar 消息保留 

在消息被确认后，Pulsar 的 Broker 会更新对应的游标。当 Topic 里面中的一条消息，被所有的订阅都确认 ack 后，才能删除这条消息。Pulsar 还允许通过设置保留时间，将消息保留更长时间，即使所有订阅已经确认消费了它们。

下图说明了如何在有 2 个订阅的主题中保留消息。订阅 A 在 M6 和订阅 B 已经消耗了 M10 之前的所有消息之前已经消耗了所有消息。这意味着 M6 之前的所有消息（灰色框中）都可以安全删除。订阅 A 仍未使用 M6 和 M9 之间的消息，无法删除它们。如果主题配置了消息保留期，则消息 M0 到 M5 将在配置的时间段内保持不变，即使 A 和 B 已经确认消费了它们。

图片

在消息保留策略中，Pulsar 还支持消息生存时间（TTL）。如果消息未在配置的 TTL 时间段内被任何消费者使用，则消息将自动标记为已确认。消息保留期消息 TTL 之间的区别在于：消息保留期作用于标记为已确认并设置为已删除的消息，而 TTL 作用于未 ack 的消息。上面的图例中说明了 Pulsar 中的 TTL。例如，如果订阅 B 没有活动消费者，则在配置的 TTL 时间段过后，消息 M10 将自动标记为已确认，即使没有消费者实际读取该消息。

四、为什么 Pulsar 更适合 IoT 场景

4.1 海量 Topic

Pulsar 计算存储分离的架构，使得 Pulsar 可以支持百万级别 Topic 数量的扩展，同时还能一直保持良好的性能。

Topic 的伸缩性取决于它的内部组织和存储方式。Pulsar 的数据保存在 BookKeeper 服务器上，处于写状态的不同 Topic 的消息，在内存中排序，最终聚合保存到大文件中，在 Bookie 中需要更少的文件句柄。另一方面 Bookie 的 IO 更少依赖于文件系统的 Pagecache，Pulsar 也因此能够支持大量的主题。

这是一个极大的提升，相比之下，Kafka 计算存储未分离，Topic 多了之后会影响其顺序 IO，性能会出现比较严重的下降。

IoT 场景的 Topic 数量是数以亿计的，Pulsar 能支持海量 Topic 的能力恰好满足了 IoT 场景的需求

4.2 租户隔离

Pulsar 通过租户和命名空间这两个关键概念支持多租户，Pulsar 的多租户性质主要体现在 topic 的 URL 中

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

多租户是 IoT 场景的一个基本需求，Pulsar 通过在 topic 中附加命名空间达到了租户隔离的效果。

4.3 消息 TTL

Pulsar 可以给未被确认的消息设置存活时长（TTL），虽然 TTL 的设置是针对整个 namespace 起效的，无法针对单个 Topic，但可以满足 IoT 场景下不同用户个性化指定 TTL 的要求

五、参考文献  

​​https://streaml.io/blog/pulsar-streaming-queuing​​

​​​​https://pulsar.apache.org/zh-CN/​​

物联网平台产品介绍详情：​​https://www.aliyun.com/product/iot/iot_instc_public_cn​​

阿里云物联网平台客户交流群