你可能并不懂 Apache Pulsar 的消息存储模型

背景svg

Apache Pulsar 系列第一篇文章为读者们详细解释了 Pulsar 的消息保留和过时策略，本文是系列第二篇，主要从 Pulsar 设计的原理以及在 BookKeeper 中如何存储作一个梳理。函数

在社区中，咱们常常能够看到用户有关 Backlog，storage size 和 retention 等策略的困惑，比较常见的一些问题，诸如：微服务

• 我没有设置 Retention 策略，为何经过 topics stats 能够查看到 storage size 远大于 backlog size？
• 个人 msg backlog size 很小，可是 storage size 确一直在增加？

• …性能

Pulsar 的消息模型

首先，咱们先来看一下 Pulsar 的消息模型

如上图所示，Pulsar 提供了最基本的 pub-sub 的处理模型。

Producer

首先 Producer 端生产消息，将消息以 append 的形式追加到 Topic 中，这里具体分发到哪个 Topic 中，根据消息是否设置了 msg key 会有所不一样。

• 设置了 msg key，消息会基于 key 作 hash，将消息分发到不一样的 partitions 中

• 未设置 msg key，消息会以 round robin 的形式，分发到不一样的 partitions 中

在消息分发的模型中，Pulsar 与 Kafka 相似。

Consumer

在 Consumer 以外，Pulsar 抽象了一层订阅层，用于订阅 Topic。经过订阅层的抽象，Pulsar 能够灵活的支持 Queue 和 Streaming 这两种类型的消息队列。每个 sub 均可以拿到这个 Topic 中全部数据的完整 copy，有点相似 Kafka 中的 consumer group。根据订阅类型的不一样，每个订阅下面能够有一个或者多个 Consumer 来接收消息。

目前，Pulsar 支持以下四种消息订阅模型：

• Exclusive
• Failover
• Shared

• Key_Shared

存储模型

消息在每一个 Partition Topic 的分布式日志中只存储一次

这就意味着，当 Producer 成功发送消息到 Topic 以后，这个消息只会在存储层存储一次，不管你有多少个 Subscription 订阅到这个 Topic 中，实际上操做的都是同一份数据。基于这个基础，咱们能够看到 Apache Pulsar 从上到下的层级抽象概念以下图所示：

首先第一层抽象是 Topic（Partition），用来存储 Producer 追加的 messages 信息，Topic 之下对应的是一个个的 ledger，ledger 里面又划分为一个个的分片，在一个个的分片中存储了更小粒度的 ertries，entries 中存储的是 【一条】或者 【一个 batch】 的消息。

• Tips: 在 Pulsar 中，一个 batch 在 broker 端会被看成一条消息来处理，batch 解析的具体逻辑是在 consumer 端接收消息时候去操做的。

• Node: 在 Bookkeeper 中，对数据操做的最小单元是按照 segment 这个粒度来进行操做的。

为何须要作分层抽象呢？

在这里最直白的解释其实就是，为了确保数据被在每个 bk 节点中打的足够散，分布的足够均匀。这也是分层分片架构设计的好处之一。

Ack 机制

在 Pulsar 中支持了两种 Ack 的机制，分别是单条 Ack 和批量 Ack。单条 Ack（AckIndividual）是指 Consumer 能够根据消息的 messageID 来针对某一个特定的消息进行 Ack 操做；批量 Ack（AckCumulative）是指一次 Ack 多条消息。

订阅机制

为了更好的理解 Strorage Size 以及 Backlog， 咱们首先须要去了解 Pulsar 中的订阅机制，以下图所示：

当有消息积压时，你能够经过 clear-backlog 来清除积压的消息。清除 backlog 中积压的消息是相对危险的操做，因此系统会提示你，是否确认要删除 backlog 中的消息， clear-backlog 提供了 -f(--force) 的参数来屏蔽该提示。

Producer 仍是按照追加的形式不断往 Topic 中发送消息，Consumer 端会建立一个 Subscription 去订阅这个 Topic，当成功订阅时，会初始化一个 Cursor 指向具体的消息的位置，默认状况下是 Latest。

Cursor 是用来存储一个订阅中消费的状态信息

上图中，咱们能够看到该订阅下面的 Topic 已经成功 Receive 而且 Ack 掉了 m4 这条消息。那么包含 m4 在内的全部的消息状态都会被标记为可删除的状态。在 Pulsar 中，使用 MarkDeletePosition 来标记这个位置。以后的全部消息，表明这个订阅尚未消费的消息。

随着时间的推移，假设在 AckCumulative 的场景下，上述订阅中的 Consumer 又消费了一些消息，目前 Cursor 的位置移动到了 m8 的位置，意味着 m8 以前的消息均可以进入删除状态。

假设是在 AckIndividual 的场景下，上述订阅中的 Consumer 只消费了 m7 这条消息而且发送了 Ack 请求，m5, m6 这两条消息仍然没有被成功消费，那么目前处于可删除状态的消息是 m4 以前的消息和 m7 这条消息。也就是说，在这种场景下，因为使用单条 Ack 致使 Topic 中间出现了 Ack 的空洞。

Cursor = Offset + IndevidualDeletes, Ack 会触发 Cursor 的移动，可是不会删除任何消息

随着时间的推移，在单条 Ack 的场景下，Ack 的空洞可能会本身消失，以下图所示：

上面咱们描述了，单个订阅在单条 Ack 和批量 Ack 混合的场景下，Topic 中 cursor 的移动状况。假设目前有多个 Subscription 订阅了这个 Topic，那么每个 Subscription 均可以拿到这个 Topic 中数据的完整 Copy，也就是一个 Subscription 会在这个 Topic 中初始化一个新的 Cursor， 每个 Cursor 之间消费的进度是没有交集、互不影响的，因此就可能出现下图中的状况：

在上图中，针对该 Topic，有两个订阅：Subscription-1 和 Subscription-2。Subscription-1中的 Consumer 消费掉了 m4 以前的消息，Subscription-2 中的 Consumer 消费掉了 m8 以前的消息。而 m4-m8 之间的这四条消息，虽然被 Subscription-2 消费完成，可是 Subscription-1 尚未消费完成这部分数据，因此这部分消息还不能够被删除。目前处于可删除状态的消息是 m4 以前的消息，即这个 Topic 中消费进度最慢的那个 Subscription 所消费完成的消息。那么这就会有一个问题，假设我目前 Subscription-1 掉线了，它的 Cursor 的位置一直没有变化，这就会致使这个 Topic 中的数据一直处于不可删除的状态。

针对上述场景，Pulsar 引入了 TTL 的概念，即容许用户设置 TTL 的时间，当消息到达 TTL 指定的阈值 Cursor 仍然没有移动的话，那么会触发 TTL 的机制，将 Cursor 自动向后移到指定的位置。在这里须要注意的一点是，咱们一直强调的是 TTL 会移动 Cursor 的位置，到目前为止，咱们尚未提到消息删除的概念，不要将两者混淆了。TTL 会作的只是去移动 Cursor 的位置，不会有任何跟消息删除的逻辑。

Backlog

为了更好的表述 Topic 中没有被消费的数据，Pulsar 引入了 Backlog 的概念来描述这一部分消息。Backlog 能够分为以下两种形式：

• Topic Backlog: 最慢的那个订阅的 Backlog 的集合

• Subscription Backlog: 指针对单个订阅级别的没有消费的数据的集合

以下图所示：Backlog A 属于 Topic Backlog；Backlog A 属于 Subscription-1 Backlog；Backlog B 属于 Subscription-2 的 Backlog。

随着时间的推移，Backlog 的会不断的变化，以下图所示：

在这里须要说明的一点是，这里的 backlogSize 记录的是带 batch 的消息，也就是一个 batch 会被看成一条消息来进行处理。由于在 broker 端去解析整个 batch 会给 broker 带来必定的负担，同时浪费大量的 CPU 资源，因此，具体 batch 逻辑的解析放到了 Consumer 端来进行处理。因此 Backlog 本质上记录的是上面咱们提到的 entries 的数量。

在 Pulsar 中，针对 Backlog 有两个指标，具体以下：

• msgBacklog: 记录的是全部未被 Ack 的 entries 的集合

• backlogSize：记录的是全部没有被 Ack 的消息的大小

Retention 机制

在 Apache Pulsar 中，使用了 BookKeeper 来做为存储层，容许用户将消息持久化，为了确保消息不会无限期的持久化下去，Pulsar 引入了 Retention 的机制，容许用户来配置消息持久化的策略。默认状况下，持久化的机制是关闭的，即消息被 Ack 以后，就会进入删除的逻辑。

配置 Retention 策略时，有以下两个参数能够指定：

• size：指持久化大小的阈值。0 表明不配置 Retention 大小策略，-1 表明设置的大小无限大

• time：指持久化时间的阈值。0 表明不配置 Retention 时间策略，-1 表明时间无限大

在引入 Retention 策略以后，整个 Topic 表示的视图以下所示，m0-m5 表明已经被全部订阅确认的消息而且已经超过了 Retention 策略的阈值，即这些消息正在 准备删除。注意，我这里描述的是 【准备删除】具体是否能够被删除，如今还不能肯定。

在最开始，咱们从最上层的 Topic 一步步抽象到了一条具体的 msg，（在这里为了方便描述，咱们忽略掉 batch 的概念，即一条 msg 等价于一个 entry）如今咱们再反过来把全部的概念都叠加回去。由于在 bk 中，容许操做的最小的单元是一个 segment，因此在具体的 msg（entry）级别，是没办法针对一条消息进行删除的，删除操做须要针对一个 segment 来进行操做。以下图所示：

假设 m0-m3 属于 segment3；m4-m7 属于segment2；m8-m11 属于 segment1。按照上图的描述，m0-m5 的消息均可以进行删除操做, 可是 segment 2 中包含了 m6, m7 并无达到 Retention 的阈值，因此 segment 目前还不能够被删除。

Storage Size

为了更方便的表述当前消息占用的存储空间的大小，Pulsar 引入了 storageSize 来描述整个概念。以下图所示：当 backlog B 与 storage Size 标识的消息相同时，backlogSize 等价于 storageSize。

当因为引入单条 Ack，Retention 策略以及 Bookkeeper 基于 segment 删除的设定，那么颇有可能形成 Storage Size 大于 backlog Size 的场景，以下图所示：

总结

1. 消息在每一个 Partition Topic 的分布式日志中只会存储一次

2. Cursor 是用来存储一个订阅下 Consumer 的消费状态的

3. Cursor 等价于 offset（kafka）+ individualDeletes

4. Ack 会去更新 Topic 中 Cursor 的位置

5. 当某条消息被全部订阅者都 Ack 以后，这条消息进入【能够被删除】的状态

6. 全部没有被确认的消息会一直保存在 Subscription backlog 中

7. TTL 能够经过设定一个时间阈值来自动更新 Cursor 的位置

8. Retention 策略是用来操做那些被 Ack 以后的消息应该怎么处理

9. 消息的删除是以 segment 为单位的，而不是 entry。

《生存仍是毁灭？一文带你看懂 Pulsar 的消息保留和过时策略》

《一天，把 Pulsar 客户端的性能提高3倍+！》

《超有料！万字详解腾讯微服务平台 TSF 的敏捷开发流程》

扫描下方二维码关注本公众号，

了解更多微服务、消息队列的相关信息！

解锁超多鹅厂周边！