对于 Apache Pulsar

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

作者介绍

本文作者：鲍明宇，腾讯 TEG 数据平台部高级工程师，Apache Pulsar Contributor，热衷于开源技术，在音讯队列畛域有丰盛教训，目前致力于 Pulsar 的落地和推广。

腾讯数据平台部 MQ 团队对 Pulsar 做了深刻调研以及大量的性能和稳定性方面优化，目前曾经在 TDbank 落地上线。本文是 Pulsar 技术系列中的一篇，次要简略梳理了 Pulsar 音讯存储与 BookKeeper 存储文件的清理机制。其中，BookKeeper 能够了解为一个 NoSQL 的存储系统，默认应用 RocksDB 存储索引数据。

Pulsar 音讯存储

Pulsar 的音讯存储在 BookKeeper 中，BookKeeper 是一个胖客户的零碎，客户端局部称为 BookKeeper，服务器端集群中的每个存储节点称为 bookie。Pulsar 零碎的 broker 作为 BookKeeper 存储系统的客户端，通过 BookKeeper 提供的客户端 SDK 将 Pulsar 的音讯存储到 bookies 集群中。

Pulsar 中的每个 topic 的每个分区（非分区 topic，能够依照分区 0 了解，分区 topic 的编号是从 0 开始的），会对应一系列的 ledger，而每个 ledger 只会存储对应分区下的音讯。对于每个分区同时只会有一个 ledger 处于 open 即可写状态。

Pulsar 在生产音讯，存储音讯时，会先找到以后分区应用的 ledger，而后生成以后音讯对应的 entry ID，entry ID 在同一个 ledger 内是递增的。非批量生产的状况（producer 端能够配置这个参数，默认是批量的），一个 entry 中蕴含一条音讯。批量形式下，一个 entry 可能蕴含多条音讯。而 bookie 中只会依照 entry 维度进行写入、查找、获取。

因而，每个 Pulsar 下的音讯的 msgID 须要有四局部组成（老版本由三局部组成），别离为（ledgerID，entryID，partition-index，batch-index），其中，partition-index 在非分区 topic 的时候为 -1，batch-index 在非批量音讯的时候为 -1。

每个 ledger，当存在的时长或保留的 entry 个数超过阈值后会进行切换，同一个 partition 下的，新的音讯会存储到下一个 ledger 中。Ledger 只是一个逻辑概念，是数据的一种逻辑组装维度，并没有对应的实体。

BookKeeper 集群中的每个 bookie 节点收到音讯后，数据会分三局部进行存储解决，别离为：journal 文件、entryLog 文件、索引文件。

其中 journal 文件，entry 数据是依照 wal 形式写入的到 journal 文件中，每个 journal 文件有大小限度，当超过单个文件大小限度的时候会切换到下一个文件持续写，因为 journal 文件是实时刷盘的，所以为了进步性能, 防止相互之间的读写 IO 相互影响，倡议存储目录与存储 entrylog 的目录辨别开，并且给每个 journal 文件的存储目录独自挂载一块硬盘（倡议应用 ssd 硬盘）。journal 文件只会保留保留几个，超过配置个数的文件将会被删除。entry 存储到 journal 文件齐全是随机的，先到先写入，journal 文件是为了保障音讯不失落而设计的。

如下图所示，每个 bookie 收到减少 entry 的申请后，会依据 ledger id 映射到存储到那个 journal 目录和 entry log 目录，entry 数据会存储在对应的目录下。目前 bookie 不反对在运行过程中变更存储目录（应用过程中，减少或缩小目录会导致局部的数据查找不到）。

如下图所示，bookie 收到 entry 写入申请后，写入 journal 文件的同时，也会保留到 write cache 中，write cache 分为两局部，一部分是正在写入的 write cache，一部分是正在正在刷盘的局部，两局部交替应用。

write cache 中有索引数据结构，能够通过索引查找到对应的 entry，write cache 中的索引是内存级别的，基于 bookie 本人定义的 ConcurrentLongLongPairHashMap 构造实现。

另外，每个 entorylog 的存储目录，会对应一个 SingleDirectoryDbLedgerStorage 类实例对象，而每个 SingleDirectoryDbLedgerStorage 对象外面会有一个基于 RocksDB 实现的索引构造，通过这个索引能够疾速的查到每个 entry 存储在哪个 entrylog 文件中。每个 write cache 在减少 entry 的时候会进行排序解决，在同一个 write cache，同一个 ledger 下的数据是相邻有序的，这样在 write cache 中的数据 flush 到 entrylog 文件时，使得写入到 entrylog 文件中的数据是部分有序的，这样的设计可能极大的进步后续的读取效率。

SingleDirectoryDbLedgerStorage 中的索引数据也会随着 entry 的刷盘而刷盘到索引文件中。在 bookie 宕机重启时，能够通过 journal 文件和 entry log 文件还原数据，保证数据不失落。

Pulsar consumer 在生产数据的时候，做了多层的缓存减速解决，如下图所示：

获取数据的程序如下：

• 在 broker 端的 entry cache 中获取，如果没有在持续；
• 在 bookie 的 write cache 正在写的这部分中获取，如果没有则持续；
• 在 bookie 的 write cache 正在刷盘的这部分中获取，如果没有则持续；
• 从 bookie 的 read cache 中获取，如果没有则持续；
• 通过索引读取磁盘上的 entry log 文件。

下面每一步，如果能获取到数据，都会间接返回，跳过前面的步骤。如果是从磁盘文件中获取的数据，会在返回的时候将数据存储到 read cache 中，另外如果是读取磁盘的操作，会多读取一部分磁盘上的时候，因为存储的时候有部分有序的解决，获取相邻数据的概率十分大，这种解决的话会极大的进步后续获取数据的效率。

咱们在应用的过程中，应尽量避免或缩小呈现生产过老数据即触发读取磁盘文件中的音讯的场景，免得对整体零碎的性能造成影响。

BookKeeper 的 GC 机制

BookKeeper 中的每个 bookie 都会周期的进行数据清理操作，默认 15 分钟查看解决一次，清理的次要流程如下：

1. 清理 bookie 存储的 ledger id（bookie 内存储的 ledger id 与 zk 下面存储的 ledger id 做比拟，如果 zk 下面没有则删除 bookie 中存储的 ledger id）；
2. 统计每个 entry log 中存活的 entry 占比，以后 entry log 存活的 ledger 个数为 0 时删除这个 entry log；
3. 依据 entry log 的元数据信息，清理 entry log 文件（当 entry log 蕴含的所有 ledger id 全副生效时删除）；
4. 压缩 entry log 文件 , 别离在以后 entry log 文件下存活的 entry 比例在 0.5- 默认周期 1 天 (major gc) 或比例 0.2- 默认周期 1 个小时 (minor gc) 的时候，Compaction entry log 文件，将老的文件中存活的 entry 转移新的文件中，而后将老的 entry log 文件删除，单次的 GC 如果解决的 entry log 文件比拟大的时候可能耗时比拟长。

通过下面的流程，咱们能够理解 bookie 在清理 entrylog 文件时的大体流程。

须要特地阐明的是，ledger 是否是能够删除的，齐全是客户端的触发的，在 Pulsar 中是 broker 触发的。

broker 端有周期的解决线程（默认 2 分钟），清理曾经生产过的音讯所在的 ledger 机制，获取 topic 中蕴含的 cursor 最初确认的音讯，将这个 topic 蕴含的 ledger 列表中，在这个 id 之前的（留神不蕴含以后的 ledger id）全副删除（包含 zk 中的元数据，同时告诉 bookie 删除对应的 ledger）。

经营中遇到的问题剖析

在使用的过程中咱们屡次遇到了 bookie 磁盘空间有余的场景，bookie 中存储了大量的 entry log 文件。比拟典型的起因次要有如下两个。

起因一 ：

生产音讯过于扩散，例如，举个极其的场景，1w 个 topic，每个 topic 生产一条，1w 个 topic 程序生产。这样每个 topic 对应的 ledger 短时间内不会因为时长或者存储大小进行切换，active 状态的 ledger id 扩散在大量的 entry log 文件中。这些 entry log 文件是不能删除或者及时压缩的。

如果遇到这种场景，能够通过重启，强制 ledger 进行切换进行解决。当然如果这个时候生产进行没有跟上，生产的 last ack 地位所在的 ledger 也是处于 active 状态的，不能进行删除。

起因二 ：

GC 工夫过程，如果现存的 enrylog 文件比拟多，且大量合乎 minor 或 major gc 阈值，这样，单次的 minor gc 或者 major gc 工夫过长，在这段时间内是不能清理过期的 entry log 文件。

这是因为单次清理流程的程序执行导致的，只有上次一轮执行完，才会执行下一次。目前，这块也在提优化流程，防止子流程执行实现过长，对整体产生影响。

小结

本文首先，介绍了 Pulsar 音讯的存储组织模式，存储流程和音讯的获取过程。其次，对单个 bookie 的 GC 流程做了详尽的阐明。在 Pulsar 的应用过程中，应该尽量避免生产过旧的历史数据即须要读取磁盘获取数据的场景。

在运维 bookie 的过程中，是不能在运行过程中调整存储目录的个数的，在部署时须要对容量进行充沛的评估。如果须要在经营的过程中进行调整时，须要对单个的 bookie 节点进行扩缩容解决。

相干浏览

• 博文举荐|多图详解 Apache Pulsar 音讯存储模型
• 博文举荐|Pulsar 存储空间不开释的问题剖析与解决办法

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