标签：bookie fragment ledger broker How Apache Pulsar Works BookKeeper

标题：Understanding How Apache Pulsar Works
原文：https://jack-vanlightly.com/blog/2018/10/2/understanding-how-apache-pulsar-works
时间：2018-10-03

声明

我感兴趣的主要特性如下：

保证不会丢失消息（如果采用了建议的配置，并且您的整个数据中心不会烧毁）
强排序保证
可预测的读写延迟

Apache Pulsar选择一致性而非可用性，其姊妹项目BookKeeper以及ZooKeeper也是如此：尽一切努力使其具有强一致性。

我们将看到Pulsar的设计，看看这些说法是否有效。在下一篇文章中，我们将测试该设计的实现。我不会在这篇文章中讨论地理复制，我们将改天再讨论，当前我们只关注单个集群。

多层抽象

Apache Pulsar具有主题和订阅的高级概念，最底层的数据存储在二进制文件中，这些二进制文件将分布在多台服务器上的多个主题的数据交织在一起。中间是无数的细节和活动部件。我个人觉得，如果我把Pulsar架构分为不同的抽象层，那么我更容易理解它，所以这就是我在本文中要做的。

图1 分层和抽象

第1层-主题、订阅和游标

这篇文章不是关于如何使用Apache Pulsar构建消息架构的。我们只介绍主题、订阅和游标的基础知识，但不会深入介绍Pulsar所支持的更广泛的消息传递模式。

图2 主题和订阅

消息存储在主题中。从逻辑上讲，主题是一个日志结构，每条消息都有一个偏移量。Apache Pulsar使用术语游标来跟踪偏移量。生产者将信息发送到给定的主题，Pulsar保证一旦消息被确认，便不会丢失（除了一些超级糟糕的灾难或配置）。

生产者可以是以下类型：

Shared：多个生产者可以发布消息到同一主题。
Exclusive：对于一个给定的主题，一次只能有一个生产者向其发布消息，如果一个主题上已经有生产者，那么其他连接到该主题的生产者将会失败。
WaitForExclusive：与Exclusive类似，区别在于当主题上有生产者之后，其他生产者会等待直到主题变成可生产状态。
ExclusiveWithFencing

消费者通过订阅消费某个主题的消息。订阅是一个逻辑实体，它跟踪光标（当前消费偏移量），并根据订阅类型提供一些额外的保证：

Exclusive Subscription：一次只有一个消费者可以通过订阅消费主题。
Shared Subscription：多个竞争关系的消费者可以通过同一订阅同时消费该主题。
Key Shared Subscription：多个消费者可以阅读主题，但不是相互竞争的关系，而是在他们之间划分key空间。这意味着任何给定key的消息总是传递给同一消费者。
Fail-Over Subscription：面向消费者的主动/备份模式，如果活跃的消费者死亡，则备份消费者将接管，但不会同时存在两个活跃的消费者。

Pub-Sub模型或Queuing模型
在Pulsar中，你可以灵活使用不同订阅。如果你想实现传统的“fan-out pub-sub”语义，可以为每个消费者指定唯一的exclusive类型订阅；如果你想实现“消息队列”语义，可以让多个消费者共享相同的订阅（类型为shared、failover、key_shared），如果你想同时实现这些语义，可以将exclusive订阅和其他类型的订阅结合起来。

一个主题可以有多个关联的订阅。订阅不包含数据，只包含元数据和游标。

创建订阅时，会创建一个关联的游标来记录上次消费的位置。

Pulsar提供了队列和日志语义，允许消费者将Pulsar主题视为在消费者确认后删除消息的队列，或者消费者可以在需要时回放游标的日志。这两种语义的底层都是以日志作为存储模型。

如果某个主题没有设置数据保留策略（通过其名称空间），那么一旦主题上关联订阅的所有游标都超过其偏移量，消息就会被删除。也就是说，该消息已在关联到该主题的所有订阅上得到确认。

但是，如果主题设置了数据保留策略，则消息一旦违反策略的限制（例如数据大小、数据保存时间等）就会被删除。

也可以发送带过期时间的消息。如果这些消息在TTL时间内未收到确认，则会被自动确认（注意不是删除）。这意味着消息可能在消费者消费之前被删除。过期删除只适用于未确认的消息，因此更符合排队语义。

TTL应用于每个订阅上，这意味着“删除”是一种逻辑删除（过期后自动确认）。实际删除将根据其他订阅情况和数据保留策略在后面进行。

译注：关于TTL的更多信息可以参考这里。

消费者可以选择逐一确认或累积确认。累积确认带来更好的吞吐量，但会在消费失败后引入重复的消息处理。但是，由于确认是基于偏移量的，因此共享订阅不能使用累积确认。消费者可以使用批量确认API，更少的RPC调用次数实现相同数量的确认，这可以提高共享订阅的吞吐量。

最后，还有一些与Kafka主题类似的分区主题。不同之处在于，Pulsar中的分区也是主题。就像Kafka一样，生产者可以采用多种算法发送消息，例如轮询算法、哈希算法或手动指定分区。

第2层-逻辑存储模型

现在Apache BookKeeper进场。我将在Apache Pulsar的上下文中讨论BookKeeper，尽管BookKeeper是一个通用的日志存储解决方案。

首先，BookKeeper跨节点存储数据。每个BookKeeper节点称为bookie。其次，Pulsar和BookKeeper都使用Apache Zookeeper来存储元数据和监视节点健康状况。

一个topic实际上是一系列ledger。一个ledger本身就是一个日志。因此，一个父日志（topic）是由一系列子日志（ledger）组成。

Topic由Ledger组成，ledger由entry（一条消息或一批消息）组成。Ledger一旦关闭，就不可更改。Ledger是作为一个整体删除的，也就是说，我们不能单独删除entry，而只能作为一个整体删除ledger。

Ledger本身也被分解成fragments。Fragments是BookKeeper集群中最小分布单元。

Topic是一个Pulsar中的概念。Ledger、fragment和entry是BookKeeper的概念，Pulsar了解并使用ledger和entry。

每个ledger（由一个或多个fragment组成）可以跨多个bookie进行复制，以实现冗余和提升读取性能。每个fragment都会在不同的bookie集合中进行复制（如果存在足够多的bookie）。

每个ledger有三个关键配置：

Ensemble size (E)
Write quorum size (Qw)
Ack quorum size (Qa)

这些配置应用于topic级别，然后Pulsar在BookKeeper的ledgers/fragments上进行设置。

注：Ensemble指将要被写入的bookie列表。Ensemble size指导Pulsar应该设置多大的ensemble。请注意，你将需要至少有E个bookie可供写入。默认情况下，从可用bookie列表中随机选择bookie（每个bookie会被注册到Zookeeper中）。

另外可以配置rack-awareness，方法是将bookie标记为属于特定机架。机架可以是逻辑结构（例如：云环境中的可用区）。根据机架感知策略，作为BookKeeper client的Pulsar broker将尝试从不同的机架中挑选bookie。还可以自定义bookie选取策略。

Ensemble Size（E）控制Pulsar可写入ledger的bookie数量。每个fragment可能有不同的ensemble，broker在创建fragment时选择一组bookie作为ensemble，ensemble大小始终由E确定。因此可用bookie数量需要大于等于E。

Write Quorum（Qw）是Pulsar将写入entry的实际bookie数量。它可以等于或小于E。将给定entry写入到的bookie集合称为write-set。

图6 一个fragment包含8个entry，存储在ensemble大小为3的bookie集群中，每个entry写到了3个bookie中

上图是由8个entry组成的fragment，存储在E=3的ensemble中，每个entry都被写入3个bookie（Qw=3）。

当Qw小于E时，就能实现条带化（striping）写入，即每个bookie只需要服务读/写请求的子集。条带化可以提高总吞吐量并降低延迟。

图7 Striping方式写入

上图表示条带化的写入。相当于5选3写入。

虽然从理论上讲，这可能有助于提高性能，但事实上，这对读性能有不利影响。BookKeeper尽了最大努力来确保顺序读，但现在条纹处理并不能很好地解决这个问题，我建议设置E=Qw。

这一段话不是很理解。

Ack Quorum（Qa）是一个微妙的话题。它表示必须确认写入的bookie数量，以便Pulsar broker向其客户返回确认。它也是replication factor的最低保证。有些entry可能只到达Qa，而没有到达Qw。

为了深入了解为什么Qa是replication factor最低保证，我在[这篇文章](/Users/oyld/Nutstore Files/我的坚果云/note/BookKeeper/Apache BookKeeper Insights Part 1.md)详细介绍了这一点。

实际上，您可以将Qa设置为Qa = Qw或者Qa = Qw -1，后者将减少Pulsar中的发布延迟和内存使用。

在创建新主题时或发生rollover时创建ledger。Rollover是指在以下情况下创建新出ledger：

ledger已达到大小或时间限制
一个主题的所有权（由Pulsar broker负责）发生了变化（稍后将对此进行详细介绍）。

在以下情况下会创建新的fragment：

新ledger被创建出来
bookie写入失败

当一个bookie无法提供写服务时，Pulsar broker就会创建一个新的fragment，并确保写操作能够得到Qw数量的bookie的确认。就像终结者，该过程持续到消息被持久化成功。

分段日志的扩展优势
增加新bookie并不意味着需要手动重新平衡。相反，这些新的bookie将成为新fragment的候选人。加入集群的bookie一旦创建了新的fragment或ledger，就会立即收到写请求。每个fragment都可以存储在集群中不同的bookie子集上。我们不会将主题或ledger绑定到某个bookie或一组bookie。

让我们停下来盘点一下。这是一个与Kafka截然不同并且更复杂的模型。在Kafka中，每个分区副本都被完整地存储在一个broker上（不过OSS即将实现分层存储）。分区副本由一系列段和索引文件组成。这篇博客很好地描述了这一点。

Kafka模式的优点在于它简单快捷。所有读取和写入都是顺序的。糟糕的是，单个broker必须有足够的存储来处理该副本，因此非常大的副本可能会迫使您拥有非常大的磁盘。第二个缺点是，在集群增长时重新平衡分区是必要的。这个过程很痛苦，需要良好的计划并顺利执行。

回到Pulsar+BookKeeper模型。给定主题的数据分布在多个bookie。这个topic被分成了ledger，ledger又被分成了fragment。当你需要扩展你的集群时，只需添加更多的bookie，当新的fragment在新bookie上被创建时，他们就会开始被写入。不再需要Kafka式的rebalancing。读写操作现在必须在bookie之间跳跃，这不是一件坏事。我们将看到Pulsar是如何管理这一点的。

每个Pulsar broker都需要跟踪每个主题所包含的ledger和fragment。此元数据存储在ZooKeeper中。

在存储层，我们将topic数据均匀的分布在BookKeeper集群中，避免了将主题副本耦合到特定节点的陷阱。Kafka的主题就像巧克力棒（sticks of Toblerone），Pulsar主题就像一种气体在膨胀，填满可用的空间，这避免了痛苦的rebalancing。

关于rebalance的对比图。

第2层-Pulsar Brokers和主题所有权

Pulsar broker位于抽象层的第二层。Pulsar broker是无状态的，它们与存储层分离。BookKeeper集群本身不执行复制，每个bookie只是一个跟随者，由leader告诉他该做什么——Pulsar broker充当leader角色。每个topic都由一个broker所有，这个broker负责该topic的所有读写工作。

当broker收到写操作时，它将向topic对应的Ensemble集合中的bookie发起写入，数据写入当前活跃的fragment。请记住，如果没有出现条带化写入，则每个entry ensemble与fragment ensemble相同（如图6所示）。如果出现条带化写入，则每个entry都有自己的ensemble，它是fragment ensemble的子集（如图7所示）。

在正常情况下，当前活跃ledger中只有一个fragment。一旦Qa个bookie确认了写操作，Pulsar broker将向生产者发送确认信息。

只有在所有前置消息都满足至少Qa个bookie已经确认的前提下，才能向生产者发送生产确认消息。如果对于给定的消息，bookie回复错误或根本不响应，那么broker将在新的bookie集合（不包括出问题的bookie）上创建一个新的fragment。

图8 单个代理处理一个topic的所有读写操作

请注意，broker只会等待Qa个bookie的确认。

读操作也要经过所有者broker。作为给定topic的单一入口点，broker知道哪些消息已安全地保存到了BookKeeper中（通过LAC）。它只需要从一个bookie上读取数据。我们将在第3层中看到它如何使用缓存来提供读服务，而不是将读取请求发送给BookKeeper。

图9 读数据只需要访问一台bookie

Broker的健康状况由ZooKeeper监控。当broker失败或不可用时，所有权发生变化。一个新的broker成为主题所有者，所有客户端现在都被引导到这台新的broker进行读写。

BookKeeper有一个非常重要的功能，叫做Fencing。Fencing允许BookKeeper保证同一时刻只有一个writer（Pulsar broker）可以写入ledger。

Last Added Confirmed (LAC)
LAC是一个ledger的提交索引。任何读取都不应超过此位置，否则会产生脏读。超出LAC之外的数据得不到一致性和正确性保证，因此任何读取都不应超过它。LAC之前的所有消息都已经被Qa台bookie确认了。LAC是作为entry的元数据保存在每一个entry中的。

Fencing流程如下：

目前拥有topic X所有权的broker（B1）被视为已死亡或不可用（通过ZooKeeper）。
另一个broker（B2）将topic X当前ledger的状态从OPEN更新为IN_RECOVERY。
B2向ledger当前fragment的所有bookie发送fencing LAC读取请求，并等待(Qw-Qa)+1个响应。一旦收到这个数量的回复，ledger现在就被围住了（fenced）。旧的broker如果仍然还活着，就不能再进行写操作，因为它将无法获得Qa确认（broker写操作会收到bookie的fencing异常响应）。

之所以要等待(Qw-Qa)+1个响应，就是确保旧broker的写请求无法得到Qa个响应。
B2收到LAC最大值，然后从LAC+1开始向后读取消息来执行恢复操作。它需要确保从LAC+1那一刻起（可能还没有向broker回复确认消息）之后的所有entry都被复制到了Qw个bookie上。一旦B2无法读取和复制更多entry，ledger就已经完全恢复了。
B2将ledger的状态更改为CLOSED。
B2打开一个新的ledger，现在可以接受对topic的写入了。

这种架构的好处在于，通过让leader（broker）没有状态，脑裂可以被BookKeeper的fencing功能轻松处理。没有脑裂，没有分歧，没有数据丢失。

Fencing过程相当于选主过程。

第2层-Cursor Tracking

每个订阅存储一个游标。游标是日志中的当前偏移量。订阅将游标存储在BookKeeper的ledger中。这使得游标跟踪和主题一样具有可伸缩性。

第3层-Bookie Storage

Ledger和fragment是在ZooKeeper中维护和跟踪的逻辑结构。从物理上讲，数据不会存储在与ledger和fragment对应的文件中。BookKeeper中存储的实际实现是可插拔的，Pulsar默认使用名为DbLedgerStorage的存储实现（index数据存储在RocksDB中）。

这里说明ledger和fragment是逻辑上的概念，并不是一个ledger对应一个文件。属于多个topic的ledger会交错保存到同一个物理文件中，即entry log file。

当向bookie写入信息时，首先将该消息写入journal file。这是一个预写日志（WAL），它可以帮助BookKeeper避免在发生故障时丢失数据。这与关系数据库实现持久性保证的机制相同。

写入的数据也会被写入write cache中。write cache会累积写操作，并定期对它们进行排序，并将其刷新到entry log files中。对写入进行排序，以便将同一ledger的条目放在一起，从而提高读取性能。如果entry是按严格的时间顺序写入的，那么读取将不会受益于磁盘上的顺序布局。通过聚合和排序，我们在ledger级别实现了时间顺序，这正是我们所关心的。

写缓存还将条目写入RocksDB，后者存储每个entry的位置索引。它只是将(ledgerId, entryId)映射到(entryLogId, 文件中的偏移量)。

读操作首先尝试命中write cache，因为write cache包含最新消息。如果write cache未命中，则会尝试命中read cache。如果第二次缓存未命中，则会在RocksDB中查找entry的位置，然后在正确的entry log file中读取该entry。它执行预读并更新read cache，以便后续请求更有可能获得缓存命中。这两层缓存意味着绝大多数读操作通常发生在内存中。

BookKeeper允许您将磁盘IO与读写隔离开来。写入操作都是按顺序写入journal file的，journal file可以存储在专用磁盘上，并分组提交，以获得更大的吞吐量。写入journal file之后，从writer的角度来看，没有产生其他磁盘同步IO，数据只是写入内存缓冲区。

异步操作发生在后台线程，write cache中的数据会批量写入entry log file和RocksDB中，它们通常使用共享磁盘。因此，一个磁盘用于同步写入（journal file），另一个磁盘用于优化后的异步写入和所有读取操作。

读数据方面，read cache或entry log files+RocksDB将为读取提供数据。

还要考虑到写操作会使入口网络带宽饱和，读操作会使出口网络带宽饱和，但它们不会相互影响。

这在磁盘和网络级别上实现了优雅的读写分离。

第3层-Pulsar Broker Caching

每个topic都有一个充当所有者的broker，所有的读写都要通过这个broker，这提供了许多好处。

首先，broker可以将日志头缓存在内存中，这意味着broker可以自己提供尾部读取服务，而不需要依赖BookKeeper。这样就避免网络往返的开销，也避免了在bookie上可能的磁盘读取操作。

Broker还知道Last Add Confirmed（LAC）entry的id。它可以跟踪哪条消息是最后一条安全持久化的消息。

当broker在其缓存中没有消息时，它将从该消息对应fragment所属ensemble bookie中的一个bookie节点请求数据。这意味着尾部读取和追赶型读取在读性能上存在很大差异。尾部读取可以通过broker上的内存提供服务，而如果写缓存和读缓存都没有数据，则追赶读取可能需要额外的网络往返开销和多次磁盘读取开销。

因此，我们从高层次上讨论了消息的逻辑和物理表示，以及Pulsar集群中的不同参与者及其相互之间的关系。还有很多细节尚未涉及，但我们将把它作为练习留到下一天。

接下来，我们将介绍Apache Pulsar集群如何确保在节点故障后充分复制消息。

Bookie手动/自动恢复

当一台bookie失效时，该bookie上的所有fragment都无法被充分复制（under replicated）。恢复过程是重新复制fragment的过程，以确保每个ledger保持复制系数Qw。

这种恢复机制不应与fencing过程中的Ledger Recovery步骤混淆，Ledger Recovery是fencing复制协议的一部分，使一个broker可以安全地关闭另一个broker的ledger。

手动/自动恢复不是BookKeeper复制协议的一部分，而是在其外部，作为一种异步修复机制存在。

有两种类型的恢复：手动或自动。二者的重新复制协议相同，但自动恢复使用内置的失败节点检测机制，注册要执行的重新复制任务。手动过程需要人工干预。

我们将关注自动恢复模式。

自动恢复可以在一组专用服务器上运行，也可以由bookie上的AutoRecoveryMain进程来完成。其中一个自动恢复进程被选为Auditor。Auditor的职责是发现被宕机的bookie，然后执行以下操作：

阅读ZK的完整ledger列表，并找到失败的bookie所拥有的ledger。
对于每个ledger，它将在ZooKeeper中的/underreplicated路径下创建一个rereplication任务。

如果Auditor节点失效，那么另一个节点将被提升为Auditor。Auditor是AutoRecoveryMain进程中的一个线程。

AutoRecoveryMain进程中还有另外一个线程负责运行复制任务，每一个worker通过监控/underreplicated路径来执行任务。

当看到任务时，它会尝试锁定它。如果无法获得锁，它将检查下一个任务（相当于一个worker负责恢复一个ledger）。

如果它确实获得了锁，那么：

扫描ledger，查找其本地bookie不存在的fragment
对于每个匹配的fragment，它将数据从另一个bookie复制到自己的bookie，在ZooKeeper上更新ensemble，fragment被标记为完全复制。

如果ledger中有剩余的未充分复制的fragment，则会释放锁。如果所有fragment都已完全复制，则任务将删除/underreplicated中的节点。

如果一个fragment没有结束entry id，那么复制任务将等待并再次检查，如果该fragment仍然没有结束entry id，则在rereplicating该片段之前，它将隔离该ledger。

因此，在自动恢复模式下，Pulsar集群能够在存储层故障时自我修复。管理员必须确保部署了适当数量的bookie。

ZooKeeper

Pulsar和BookKeeper都依赖ZooKeeper。如果一个Pulsar节点与所有ZooKeeper节点都不通，那么它将停止接受读写操作，并重新启动自身。这是一种预防措施，以确保集群不会进入不一致的状态。

这确实意味着，如果ZooKeeper关闭，一切都将不可用，所有Pulsar节点缓存都将被清除。因此，在恢复服务后，理论上可能会出现延迟峰值，因为所有读取都会转到BookKeeper。

小结

每个topic都有一个所有者broker
每个topic在逻辑上被分解为ledger、fragment和entry
fragment分布在整个bookie集群中，给定的topic与给定的bookie不存在耦合
fragment可以在多个bookie上条纹分布
当Pulsar broker失败时，该broker的topic所有权将转移给另一个broker，Fencing可以避免两个可能认为自己是所有者的broker同时向当前topic ledger写数据
当一个bookie失败时，自动恢复（如果启用）机制将自动向另一台bookie执行复制操作，如果禁用，可以启动手动恢复过程
Broker缓存数据，使其能够非常高效地提供尾部读服务
Bookie使用journal为失败提供保证，journal可用于恢复发生故障时尚未写入entry log file的数据
所有topic的entry在entry log file中交错存储，用于查找的索引保存在RocksDB中
Bookie读取流程如下：write cache -> read cache -> log entry files
Bookie可以通过分离journal files、log entry files和RocksDB的磁盘实现读写分离
ZooKeeper存储Pulsar和BookKeeper的所有元数据，如果ZooKeeper不可用，Pulsar便不可用
存储层可以单独扩展，如果存储是瓶颈，那么只需添加更多bookie，他们就可以开始加载，而无需数据重平衡

关于数据丢失

如前所述，Qa是复制因子。因此，要想丢失数据，就必须失去Qa个bookie。这就是为什么Qa设置成1是不安全的。

当您使用基于超过半数仲裁的系统（例如acks=all且min insync replicas=2且rep-factor=3的Apache Kafka）时，您的最低保证复制系数就是超过半数。因此，当复制因子为3时，最低保证为2。所以这和BookKeeper没什么不同。与BookKeeper不同的是，你可以明确地控制最低法定人数。

由于BookKeeper在确认写操作之前先将数据写入WAL，并通过fsyncs写入磁盘，因此Pulsar在整个集群突然发生故障的情况下（例如断电事件）不会丢失数据。这使Pulsar比Kafka更安全。

关于可用性

Pulsar比RabbitMQ或Kafka等其他系统具有更好的写可用性，因为每个topic都是动态的。一个topic是一个分段的日志，由ledger组成，可以自由地从失败的bookie中移除。整个bookie ensemble都允许宕机，Pulsar只需关闭当前的ledger，并在一组正常运作的bookie上打开一个新的ledger，就可以继续运行下去。需要BookKeeper提供的读取将受到影响，但只要fragment对应的bookie ensemble中有一个bookie可用，读取也可以继续。

使ack quorum为1是不安全的，它不仅使数据没有冗余，还可能会阻止ledger recovery。Ledger recovery需要不断读取，直到找到最后提交的entry。但如果Qa为1，并且有一个bookie出现故障，它无法确定该bookie是否拥有最新entry，因此恢复过程将暂停。

总结

Apache Pulsar在协议和存储模型方面比Apache Kafka复杂得多。

Pulsar的两个突出特征是：

将broker与存储分离，再加上BookKeepers的fencing功能，优雅地避免了可能导致数据丢失的脑裂情况。
将topic分解成ledger和fragment，并将它们分布在一个集群中，可以让Pulsar集群轻松扩展，新数据自动开始写入新的bookies，无需数据重平衡。

此外，我甚至还没有提到地理复制和分层存储，它们也是惊人的功能。

我觉得Pulsar和BookKeeper是下一代数据流系统的一部分。他们的协议经过深思熟虑，相当优雅。但随着复杂性的增加，出现漏洞的风险也随之增加。在下一篇文章中，我们将开始对Apache Pulsar集群进行混沌测试，看看我们是否能够识别协议中的弱点，以及任何实现错误或异常。

我写的关于Pulsar的文章：

我写的关于Apache BookKeeper的文章

标签：bookie,fragment,ledger,broker,How,Apache,Pulsar,Works,BookKeeper
From： https://www.cnblogs.com/oyld/p/16840645.html

Understanding How Apache Pulsar Works

声明