一、前言

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​​

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