编者荐语：
原文作者冉小龙，首发于公众号“腾讯云中间件”，公布已取得原帐号受权。如需转载，请返回联系。本文次要为大家介绍 Apache Pulsar 的音讯和存储模型，供大家参考。

对于 Apache Pulsar

Apache Pulsar 是 Apache 软件基金会顶级我的项目，是下一代云原生分布式音讯流平台，集音讯、存储、轻量化函数式计算为一体，采纳计算与存储拆散架构设计，反对多租户、长久化存储、多机房跨区域数据复制，具备强一致性、高吞吐、低延时及高可扩展性等流数据存储个性。
GitHub 地址：http://github.com/apache/pulsar/

背景

第一篇 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。

对于作者

冉小龙，腾讯云微服务产品核心研发工程师，Apache Pulsar Committer，Apache BookKeeper Contributor

相干举荐

• Pulsar namespace 策略浅析
• Message Lifecycle：Pulsar 里的信息传递到底是什么样子
• 了解 Apache Pulsar 工作原理

点击 链接，获取 Apache Pulsar 硬核干货材料！