下一代云原生消息流平台 Apache Pulsar 消息保留和过期策略设计

导读

Apache Pulsar 是 Apache 软件基金会顶级项目，是下一代云原生分布式消息流平台，集消息、存储、轻量化函数式计算为一体，采用计算与存储分离架构设计，支持多租户、持久化存储、多机房跨区域数据复制，具有强一致性、高吞吐、低延时及高可扩展性等流数据存储特性。

作者介绍

冉小龙

腾讯云微服务产品中心研发工程师
Apache Pulsar Committer
Apache BookKeeper Contributor

默认情况下，Pulsar Broker 会对消息做如下处理：

• 当消息被 Consumer 确认之后，会立即执行删除操作。

• 对于未被确认的消息会存储到 backlog 中。

但是，很多线上的生产环境下，这种默认行为并不能满足我们的生产需求，所以，Pulsar 提供了如下配置策略来覆盖这些行为：

• Retention 策略：用户可以将 Consumer 已经确认的消息保留下来。

• TTL 策略：对于未确认的消息，用户可以通过设置 TTL 来使未确认的消息到达已经确认的状态。

上述两种策略的设置都是在 NameSpace 的级别进行设置

Retention 策略

Retention 策略的设置提供了两种方式：

• 消息的大小，默认值：defaultRetentionSizeInMB=0

• 消息被保存的时间，默认值：defaultRetentionTimeInMinutes=0

我们可以在 broker.conf 中对这两项内容进行配置也可以通过命令行的形式。上文提到过，这两种策略的设置，都是在 NameSpace 的级别进行设置，所以当我们使用命令行配置时，使用 NameSpaces 来进行配置，具体如下：

root@e6df71e544ea:/pulsar# ./bin/pulsar-admin namespaces set-retention
The following options are required: --size, -s --time, -t

Set the retention policy for a namespace
Usage: set-retention [options] tenant/namespace
  Options:
  * --size, -s
       Retention size limit (eg: 10M, 16G, 3T). 0 or less than 1MB means no
       retention and -1 means infinite size retention
  * --time, -t
       Retention time in minutes (or minutes, hours, days, weeks eg: 100m, 3h, 2d,
       5w). 0 means no retention and -1 means infinite time retention

如上所示：我们可以通过 -s 或者 -t 来指定我们需要配置的大小或者时间。

当你设置 Retention 策略之后，可以通过如下命令来查看具体的信息：

$ pulsar-admin namespaces get-retention [your tenant]/[your-namespace]
{
  "retentionTimeInMinutes": 10,
  "retentionSizeInMB": 0
}

Backlog

backlog 是未被确认消息的集合，它有一个大前提是，这些消息所在的 Topic 是被 Broker 所持久化的，在默认情况下，用户创建的 Topic 都会被持久化。换句话说，Pulsar Broker 会将所有未确认或者未处理的消息都存放到 backlog 中。

同样的，我们可以在 NameSpace 级别对 backlog 的大小进行配置。需要注意的是，对 backlog 进行配置时，我们需要明确以下两点：

• 在当前的 NameSpace 下，每一个 Topic 允许的大小是多少

• 如果超过设定的 backlog 的阈值，将会执行哪些操作

当超过设定的 backlog 的阈值，Pulsar 提供了以下三种策略供用户选择：

你可以通过 set-backlog-quota 在 NameSpace 级别对 backlog 进行配置，具体如下：

root@e6df71e544ea:/pulsar# ./bin/pulsar-admin namespaces set-backlog-quota
The following options are required: -l, --limit -p, --policy

Set a backlog quota policy for a namespace
Usage: set-backlog-quota [options] tenant/namespace
  Options:
  * -l, --limit
       Size limit (eg: 10M, 16G)
  * -p, --policy
       Retention policy to enforce when the limit is reached. Valid options are:
       [producer_request_hold, producer_exception, consumer_backlog_eviction]

如上所示，set-backlog-quota 提供了两个参数，-l 用来指定你设置 backlog 的大小，-p 用来指定，当超过你设置的 backlog 的阈值之后，Broker 将会执行的策略。

当你设置 backlog 之后，可以通过如下命令，查看相应的信息：

$ pulsar-admin namespaces get-backlog-quotas [your tenant]/[your namespace]
{
  "destination_storage": {
    "limit" : 2147483648,
    "policy" : "producer_request_hold"
  }
}

如果你期望取消 backlog 的配置，可以使用如下命令：

$ pulsar-admin namespaces remove-backlog-quota [your tenant]/[your namespace]

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

$ pulsar-admin namespaces clear-backlog [your tenant]/[your namespace]

Time To Live （TTL）

默认情况下，Pulsar 会持久化所有未被确认的消息。如果未被确认的消息有很多，这种策略会造成大量的消息被积压，导致磁盘空间增大。有些场景下，消息并不需要被持久化，用户更期望的行为是，将这些未被确认的消息直接丢弃。这种情况下，你可以通过设置 TTL 使得未被确认的消息进入到被确认的状态，当超过设定的 TTL 时间之后，配合相应的 Retention 策略将消息丢弃。

TTL 的一个典型使用场景是，当 Consumer 由于某些原因出现故障，不能正常消费消息，这时 Producer 还在不断的往 Topic 中生产消息，会造成 Topic 中有大量的未确认的消息出现，这时你可以通过设置 TTL 将这些未确认的消息变为已确认的状态。

同样的，你可以在 Namesapce 级别下，通过指定 set-message-ttl 对 TTL 进行设置，具体命令如下：

root@e6df71e544ea:/pulsar# ./bin/pulsar-admin namespaces set-message-ttl
The following option is required: --messageTTL, -ttl

Set Message TTL for a namespace
Usage: set-message-ttl [options] tenant/namespace
  Options:
  * --messageTTL, -ttl
       Message TTL in seconds
       Default: 0

如上所示，set-message-ttl 只有一个参数 -ttl，单位为秒，默认值为 0。

当你设置 TTL 策略之后，可以通过 get-message-ttl 查看相应的配置信息，具体如下：

$ pulsar-admin namespaces get-message-ttl [your tenant]/[your namespace]
60

TTL、Backlog 与 Retention 的区别和联系

在上述的描述过程中，可以发现，TTL 只去处理一件事情，将未被确认的消息变为被确认的状态，TTL 本身不会去涉及相应的删除操作，具体如下图所示：

1. 在 T1 阶段，m1-m5 这 5 条消息被确认，m6-m10 这 5 条消息未被确认
2. 在 T2 阶段，对 m6、m7、m8 这三条消息设置 TTL 策略

3. 在 T3 阶段，到达 TTL 设定的阈值，m6、m7、m8 这三条消息被确认

通过上图可以看到，对于 backlog 中未被确认的消息，当你设置 TTL 之后，会将未确认消息的状态变为确认的状态。TTL 在这里所起到的作用就是将消息的 Cursor 从 m5 移动到 m8，m6、m7、m8 这三条消息变为已确认状态。

Pulsar 是一个 multiple-subscription 的消息系统，对于 Topic 中的一条消息，只有当所有订阅者都对这条消息 ack 或者消费了，它才能被删除。

默认情况下，Pulsar Broker 会将所有未确认的消息持久化到 backlog 中。TTL 的功能是，你可以将这些未被确认的消息变为被确认的状态，而 Retention 所关注的点是，当消息处于被确认的状态时，你可以对已确认的消息进行的保留策略是什么。换句话说，backlog 是针对未确认的消息，Broker 所做的处理是什么。Retention 是针对已确认的消息，Broker 所做的保留策略是什么。

TTL 与死信队列

死信队列的相关介绍在此不做赘述。

在生产环境中，有时可能遇到质量差的数据是由于上游的原因导致的，必须由上游来解决，继续尝试处理其它的消息已经没有意义，这时候用户希望在发生错误时立即停止处理。Pulsar 中提供了一种特殊的 Topic——死信队列。

死信队列与 TTL 都可以将未确认的消息变为已确认的状态。他们之间主要的不同在于，在上图中的 T2 阶段，TTL 只是将未确认的消息变为已确认的状态，死信队列的做法是将消息丢弃到死信队列中，m6、m7、m8 这三条消息变为被确认的状态。m9、m10 这两条有效消息会正常处理，Broker 也会继续运行。之后，你可以从死信队列中检查无效消息，并根据需要忽略或修复并重新处理。用户可以根据自己的需求来确定未确认的消息是通过 TTL 的形式将其变为确认状态还是通过死信队列的方式来实现，依据的主要标准就是看你需不要处理消费不了的消息。

使用问题

• 场景一：

启动 Producer 往 Broker 发送消息，设置了 TTL ，没有启动 Consumer，同时设置了 Retention 策略为半小时，到达 Retention 的阈值之后，发现设置 TTL 的消息并没有被移除，这是为什么呢？

在上述场景中，有一个问题需要注意，没有启动 Consumer，在上面我们说到，TTL 是将消息设置的 Cousor 向前移动，如果没有启动 Consumer，相当于 Cousor 没有被初始化，也就是如果没有 Consumer，你就没有必要去设置 TTL。

• 场景二：

我设置了Retention 策略，但是到达了 Retention 的阈值，Topic 中的数据并没有被删除掉，这是为什么呢？

这个是 Pulsar 内部的一个实现机制，在 Pulsar 中 Topic 是一个逻辑的概念，一个 Topic 对应一个 manage ledger，当你写数据的时候，实际上是将数据写到了 ledger 中，还记得在之前很多文章中提到的有关 Pulsar 设计的一个核心所在：在 Pulsar 中，所有的操作都是异步的，所以当 Retention 到达指定阈值之后，是否删除对应 ledger 中的数据，这个操作也是异步的。delete 的操作是不会对当前 active 的 ledger 执行的。只有当数据写满了当前的 ledger ，ledger 发生切换时，才会去真正的执行 retention 策略。

如果想要强制执行，可以使用 pulsar-admin 将当前的 ledger 强制卸载，迫使其发生 ledger 的切换。

参考阅读

技术原创及架构实践文章，欢迎通过公众号菜单「联系我们」进行投稿。

高可用架构

改变互联网的构建方式