本文为《Pulsar Topic Compaction》中文翻译版本。 
原文链接：https://www.splunk.com/en_us/blog/it/pulsar-topic-compaction.html#:~:text=Topic%20compaction%20is%20a%20process,the%20users%20who%20want%20it.ww.splunk.com/en_us/blog/it/simple-event-processing-with-apache-pulsar-functions.html

阅读本文需要大约 4 分钟。

用户订阅 Pulsar 中的 topic，通常只关注消息的最新键值。通读所有积压消息的最新键值需要用户读取大量消息，而这些消息将会被后续消息阻塞。这样的操作不仅浪费时间，还浪费资源。

为了解决这一问题，Apache Pulsar 2.0 引入了 Topic 压缩。Topic 压缩进程在 Pulsar broker 中运行，用于为 topic 中的每个键生成最新值的快照。和其他类似的消息系统相比，Pulsar Topic 压缩没有破坏性。对于需要原始积压消息的用户来说，这些消息仍然有用。Pulsar 支持通过 REST 接口手动触发 Topic 压缩，因此用户可以完全自行控制 Topic 压缩的时间。

Topic 压缩进程读取 topic 中积压的全部消息后，仅保留压缩消息的最新键值。如果键的最新有效负载为空，则认为该键也为空，因此不为其保存任何消息。

用例：股票价格

以股票报价为例。当股价改变时，即向 topic 写入消息，该消息包含两部分：一部分为消息有效负载中股票变化后的价格，另一部分为将股票股价变化值的正负作为消息键。

这样的 topic 有两种 consumer。一种 consumer 关注股价如何随时间改变，因而可以采用机器学习或更简单的方式为用户展示股价变化。

而另一种 consumer 则只关注最新股价，例如用于计算用户所有投资的总价值。这种 consumer 将从压缩后的积压消息中读取数据。

图 1. Pulsar Function 运行图

从压缩 Topic 中消费

想要通过读取 topic 中压缩后的积压消息来迅速获取 topic 的最新状态，客户端必须在创建 consumer 时指定 readCompacted(true)。

如果客户端想要读取 topic 中的所有积压消息，则可以指定 readCompacted(false)，或不对 readCompacted 参数进行赋值，因为默认情况为读取全部积压消息。

无论从已压缩积压消息中读取还是从所有积压消息中读取，消息 ID 都不会改变，因此压缩不会对订阅游标产生影响。例如，以上图中的 topic 为例，假设客户端从所有积压消息中进行读取（如果 topic 没有被压缩，通常会以这种方式读取），在客户端读取至序号为 3 的消息后断开一段时间，如果这时发生了压缩操作，而客户端重新连接成功，并指定从已压缩消息中进行读取，则客户端从已压缩消息中序号为 6 的消息开始读取数据。

触发压缩

目前可以通过 rest API 触发 Topic 压缩。调用命令行示例如下。例如，要在上述 topic 中启动压缩，运行以下命令：

触发压缩只会启动压缩，在压缩完成前不会发生阻塞。可以通过以下命令查看压缩进程的状态：

向压缩状态子命令添加 -w 参数，即可等待压缩完成。

Pulsar 中的 Topic 压缩受 topic 中数据保留时间的影响。Backlog 中的消息过了保留时间会被删除，这些消息也会从压缩的 backlog 中删除。如果不希望删除消息，则可以配置 topic 为无限期保留。

从 2.1 版本起，Pulsar 支持用户设定 topic 压缩阈值，当 topic 数据量到达该阈值会触发自动压缩操作。

相关阅读

 • 译文｜LogDevice 与 Apache Pulsar 之间的对比

 • 译文｜Pulsar Schema Registry

点击「阅读原文」进入 Pulsar 世界！