Pulsar 分层存储与 BookKeeper 原理

可视化演示

演示Pulsar计算存储分离架构、BookKeeper的Ledger与Entry复制模型，以及热冷分层存储的卸载与成本优化，可切换三种模式观察。

分层存储架构

想象那座智慧仓库。仓库主管最头疼的是“好地方有限”：收银台旁的快速取货区最贵但也最好用，地下冷库便宜却要坐电梯。聪明的仓库不会把三个月前的包裹堆在取货区，而是按“多久会被取走”分区域摆放。Pulsar 的分层存储，就是给消息仓库设计了同样的分区规矩。

在 Pulsar 中，处理消息的 Broker 只负责“接待和分拣”（计算），真正存放消息的是一套独立存储集群，用 BookKeeper 来管理，这就是“计算与存储分离”：收银员和仓库不再挤在同一间屋子里。需要更强的收发能力就多加 Broker，需要更大空间就多部署 Bookie 存储节点，两边各自扩容、互不干扰。

消息写入后按“年龄”自动分层：最新的一批留在最贵最快的热层，通常由 BookKeeper 的磁盘缓存承接；再过几天没被消费，就留在 BookKeeper 的常规存储（温层）；一旦账本关闭且超过配置的卸载阈值，后台的卸载任务会把这段数据搬去 S3 等对象存储，这就是冷层。越冷的数据，每 GB 单价越便宜。

这套自动分拣就像仓库里的自动叉车，不需要主管操心，时间一到旧包裹自己就搬去冷库了。于是无论消息保留多久，热区永远清爽，账单永远可控。

账本核心模型

仓库里最怕什么？丢件。为了不丢件，每一件包裹的出库入库都要记在账本上。BookKeeper 这个名字本身就是“记账人”，它的核心模型就是“账本加账页”。

BookKeeper 把消息组织成 Ledger（账本分册）和 Entry（账页条目）。一个 Ledger 是一段连续追加的日志，有唯一的编号；写入方把消息按顺序追加成 Entry，每条 Entry 有一个单调递增的编号。一个 topic 的存储会被切分成多个 Ledger，每个 Ledger 通常对应几十到几百 MB 的数据段——就像仓库把流水账按月份装订成册，而不是一本记到天荒地老。

每个 Ledger 的 Entry 不是只存在一个地方。BookKeeper 会把同一份数据分散存到多个存储节点（Bookie）上。为了容错，通常会配置：总副本数（ensemble size）3 份、写入时需要至少 write quorum 个节点参与、确认时至少 ack quorum 个节点返回成功（一般 2）。这就是“多人各抄一份账本，关键条目要至少两人确认”。

有了账本分册，阅读方可以自由地按 Ledger 顺序回放，也可以只翻某一册。账本一旦写满并关闭就不再改动，天然支持只读归档——这正是“装订好的账本不能撕页重写”的道理。

读写一致性

几个仓库管理员同时记账，最容易出乱子：新来的管理员还没抄完，就有人拿着他的账本去发货。消息系统也一样，读者如果读到“尚未被足够多副本确认”的数据，在节点故障时就会出现读到一半又没了的情况。

BookKeeper 用 LAC（Last Add Confirmed，最后确认水位）来解决这个问题。写入方每次追加 Entry，要等足够多的副本回复“我记下了”，才会更新 LAC。读取方只承诺读到 LAC 为止的数据，LAC 之后就属于“尚未确认区”，读者要么等，要么被告知读不到。这就像仓库发货前要核对盖章数，没盖满章的记录绝不会被取出仓库。

写入流程是纯追加的：先选一个 Ledger 的尾部，把 Entry 发给多个 Bookie，Bookie 落盘后再应答。acks 满足 ack quorum 后，才回给生产者“成功”。读取时按 EntryId 顺序读取，并且借助 LAC 知道该读到哪，不会读到“空中楼阁”。

靠着这条“确认水位”，BookKeeper 不让读者看到空中楼阁。即使某台 Bookie 突然断电，只要确认数达标，账本记录依然安全——别的分店还有同样的盖章记录。

存储模型对比

如果消息仓库也有“同行”，Kafka 就像一家用“一条超长传送带”管理的仓库。每个分区是一条连续队列，放进传送带的包裹依次经过所有工位，谁要取货就在传送带上捞。传送带天然顺序好、性能好，但毛病是：传送带和地盘绑死了。

在 Kafka 中，一个 Partition 是一个物理日志，固定由某个 Broker 节点的磁盘承载。分区数越多、磁盘 IO 越吃紧，Broker 就像传送带机房堆满了货物，加机器还得先把旧货搬过去（数据重平衡）。Pulsar 则不同：Topic 只是逻辑名字，真正存储的单元是很多个 Ledger，散落在整个 BookKeeper 集群里。存储节点不够？直接加 Bookie，新账本自动往新节点上放，几乎不需要搬旧数据。

两者的一致性模型也不同。Kafka 用 ISR + 高水位（HW）表示“哪些消息可以被消费”，HW 是 ISR 中所有副本日志末尾的最小值减一；Pulsar 用 LAC 表示每个 Ledger 的确认水位。一个以分区为单位做整体水位管理，一个以账本为粒度逐条确认。

这份差异决定了性格：Kafka 适合轻量、极致的吞吐和单分区有序；Pulsar 适合高并发多租户、存储弹性伸缩和长保留。就像传送带仓库适合高产出的单一流水线，而账本仓库适合五花八门、存放期长的综合物流。

保留与容灾

仓库不能无限堆货，过期的包裹要么销毁，要么归档，还要防着货架倒塌。这一节看 Pulsar 如何做“定期盘点与灾备”。

Pulsar 的保留策略有两种：按时间（如保留 7 天）或按大小（如最多 10GB），超出的历史消息会被自动删除或卸载到冷层。消息积压时不用怕——因为存储和计算分离，消费者离线多久，消息都安安静静躺在 BookKeeper 里，积压只增加存储成本，不拖垮 Broker 内存。

容灾方面，BookKeeper 有自动修复（AutoRecovery）：当某个 Bookie 节点挂掉，审计线程发现某些 Ledger 的副本数低于期望值，就会自动把缺失的副本复制到其他健康节点，就像仓库货架倒了，马上从别处分店抄一份补回来。此外，Ledger 写入达到上限会滚动（roll）到新 Ledger，旧的关闭后变成只读，方便备份和卸载。

因此 Pulsar 用户最常见的体验是：不管消费者宕多久，重新上线后消息原封不动等着它；存储空间不足时，加节点即可，系统自动把存量账本副本重新铺开。

选型与迁移

换仓库管理系统是件大事。知道自己要什么，才能决定用传送带还是账本仓库。Pulsar 的优势场景很清晰：多租户共享集群、消息保留超过几天甚至几个月、需要存储与计算独立扩容、跨地域复制。

如果团队已在使用 Kafka，迁移要讲究平滑。常见路径是“双写 + 回放”：让生产者在过渡期同时写 Kafka 和 Pulsar，消费者逐步切到 Pulsar，确认稳定后再停掉 Kafka 写入；也可以借助 Pulsar 的 KoP（Kafka-on-Pulsar）协议适配层，让现有 Kafka 客户端几乎零改动连上 Pulsar。

配置实践上，关键参数包括副本三件套（managedLedgerDefaultEnsembleSize / WriteQuorum / AckQuorum）、保留策略（retentionTimeInMinutes、retentionSizeInMB）以及卸载阈值（managedLedgerOffloadThreshold）等。运维重点则是紧盯 Bookie 磁盘使用率、LAC 是否健康、副本数是否达标，以及卸载任务是否按时执行。

选型没有银弹，但给仓库装上“自动分拣 + 灵活账本”后，长期运营和弹性扩容会从容很多。如果团队正为“消息保留 30 天就爆磁盘”发愁，Pulsar 的分层存储很可能是你那把自动叉车。

AI 应用场景

Python 演示

class TieredStorage:    """模拟 Pulsar 分层存储: 消息随年龄从热层自动沉降到温层和冷层"""     def __init__(self, hot_cost=1.0, warm_cost=0.2, cold_cost=0.03):        self.hot = []       # 每条消息为 [msg_id, age]        self.warm = []        self.cold = []        self.hot_cost = hot_cost        self.warm_cost = warm_cost        self.cold_cost = cold_cost        self.msg_id = 0     def write(self, data):        self.hot.append([self.msg_id, 0])   # 新消息写入热层        self.msg_id += 1     def tick(self):        # 所有消息年龄 +1        for bucket in (self.hot, self.warm):            for item in bucket:                item[1] += 1        # 热 -> 温: 年龄 >= 3        moved = [i for i in self.hot if i[1] >= 3]        self.warm.extend(moved)        self.hot = [i for i in self.hot if i[1] < 3]        # 温 -> 冷: 年龄 >= 10        moved = [i for i in self.warm if i[1] >= 10]        self.cold.extend(moved)        self.warm = [i for i in self.warm if i[1] < 10]     def monthly_cost(self):        return (len(self.hot) * self.hot_cost * 30 +                len(self.warm) * self.warm_cost * 30 +                len(self.cold) * self.cold_cost * 30)  def main():    ps = TieredStorage()    daily = 100    for day in range(30):        for j in range(daily):            ps.write("msg-%d-%d" % (day, j))        ps.tick()        if day in (6, 15, 29):            print("第 %2d 天: 热=%4d 温=%4d 冷=%4d 月成本约 %6.0f" %                  (day + 1, len(ps.hot), len(ps.warm), len(ps.cold), ps.monthly_cost()))    no_tier = 3000 * ps.hot_cost * 30   # 不分层成本    tiered = ps.monthly_cost()    print("不分层成本: %.0f, 分层后成本: %.0f, 节省: %.1f%%" %          (no_tier, tiered, (1 - tiered / no_tier) * 100))  if __name__ == "__main__":    main()

1. self.hot.append([self.msg_id, 0])每条新消息先进入热层，年龄为 0，对应 Pulsar 消息先落在最快的存储。
2. for bucket in (self.hot, self.warm):每 tick 一次，热层和温层的消息年龄都加 1，模拟时间流逝。
3. moved = [i for i in self.hot if i[1] >= 3]找出年龄达到 3 的消息，准备搬到温层，对应 ledger 关闭后的卸载条件。
4. self.hot = [i for i in self.hot if i[1] < 3]热层只保留年轻消息，实现“自动沉降”，旧数据不再占用昂贵空间。
5. no_tier = 3000 * ps.hot_cost * 30不分层时，3000 条消息全部按最贵单价计费，作为成本对比基线。
6. (1 - tiered / no_tier) * 100计算分层后的成本节省百分比，运行脚本会打印出 83% 左右的节省效果。

常见误区

面试题