Apache Pulsar工作原理

在本文中，我们将介绍ApachePulsar的设计，以便更好地设计故障场景。这篇文章不是为那些想了解如何使用ApachePulsar的人写的，而是为那些想了解它是如何工作的人写的。我一直在努力以一种简单易懂的方式对其架构进行清晰的概述。

主要的声明包括：

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

Apache Pulsar选择一致性而不是可用性，其姊妹项目BookKeeper和ZooKeeper也是如此。尽一切努力使其具有很强的一致性。

我们将看看Pulsar的设计，看看这些说法是否有效。

多层抽象

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

让我们沿着层层往下走。

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

这篇文章不是关于可以使用Apache Pulsar构建的消息传递体系结构的。我们将只介绍主题、订阅和游标的基本内容，但不会深入了解Pulsar支持的更广泛的消息传递模式。

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

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

• 独家订阅-一次只有一个消费者可以通过订阅阅读主题
• 共享订阅-竞争消费者可以通过同一订阅同时阅读主题。
• 故障转移订阅-用户的活动/备份模式。如果活动consumer消费者死亡，则备份将接管。但从来没有两个活跃的消费者同时存在。

一个主题可以有多个附加订阅。订阅不包含数据，只包含元数据和光标。

Pulsar提供了排队和日志语义，允许消费者将Pulsar主题视为在消费者确认后删除消息的队列，或者消费者可以在需要时回放光标的日志。下面的存储模型是相同的-一个日志。

如果未对主题（通过其命名空间）设置数据保留策略，则在附加订阅的所有游标都通过其偏移量后，将删除消息。也就是说，邮件已在附加到该主题的所有订阅上得到确认。

但是，如果存在覆盖主题的数据保留策略，则消息在通过策略边界（主题大小、主题中的时间）后将被删除。

也可以发送过期的消息。如果这些消息在未确认的情况下超过TTL，则会将其删除。这意味着可以在任何消费者有机会阅读它们之前将其删除。过期只适用于未确认的消息，因此更适合于事物的排队语义方面。

TTL分别应用于每个订阅，这意味着“删除”是一种逻辑删除。实际删除将根据其他订阅和任何数据保留策略中发生的情况在以后进行。

消费者一个接一个地或累积地确认他们的信息。累积确认将更好地提高吞吐量，但会在使用者失败后引入重复的消息处理。但是，累积确认不适用于共享订阅，因为确认基于偏移量。但是，使用者API允许批处理确认，这些批处理确认最终将得到相同数量的确认，但RPC调用更少。这可以提高共享订阅上竞争消费者的吞吐量。

最后，还有一些与Kafka主题类似的分区主题。不同的是，Pulsar中的分区也是主题。就像kafka一样，生产者可以循环发送消息，使用散列算法或显式选择分区。

这是对高级概念的旋风式介绍，我们现在将深入探讨。请记住，这不是从10000英尺的高度对Apache Pulsar的入门介绍，而是从1000英尺的高度看它是如何在下面工作的。

第2层-逻辑存储模型

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

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

一个主题实际上是一系列Ledgers。Ledger本身就是一个日志。因此，我们从一系列子日志（分类账）组成一个父日志（主题）。

Ledger附加到主题，entries（消息或消息组）附加到Ledgers。Ledgers一旦关闭，就不可更改。Ledgers是作为一个整体删除的，也就是说，我们不能删除单个条目，而是作为一个整体删除Ledgers。

Ledgers本身也被分解成碎片。片段是BookKeeper集群中最小的分布单元（根据您的观点，条带化可能会使该声明无效）。

主题是一个Pulsar概念。Ledgers, Fragments和Entries是BookKeeper的概念，尽管Pulsar了解并使用Ledgers和Entries。

每个Ledger（由一个或多个片段组成）可以跨多个BookKeeper节点（Bookies）进行复制，以实现冗余和读取性能。每个片段在不同的一组Fragment中复制（如果存在足够多的Fragment）。

每个分类账有三个关键配置：

• Ensemble大小（E）
• 写入Quorum大小（Qw）
• 确认Quorum大小（Qa）

这些配置应用于主题级别，然后Pulsar在BookKeeper Ledgers/Fragments片段上设置。

注：“Ensemble”指将被写入的实际Bookies列表。Ensemble大小是一个指示Pulsar说它应该创造多大的Ensemble。请注意，您将需要至少E个可供写入的Bookies。默认情况下，从可用Bookies列表中随机选择bookies（每个bookies在Zookeeper中注册）。

Ensemble大小（E）控制可供Pulsar写入该Ledger的Bookies的大小。每个片段可能有一个不同的Ensemble，broker将在创建fragment时选择一组Bookies，但Ensemble的大小将始终为E指示的大小。必须有足够多的Bookies可以承保E。

Write Quorum（Qw）是Pulsar将写入条目的实际Bookies数量。它可以等于或小于E。

当Qw小于E时，我们得到striping，它以这样一种方式分配读/写操作，即每个Bookie只需要为读/写请求的子集提供服务。条带化可以增加总吞吐量并降低延迟。

Ack Quorum（Qa）是必须确认写入的Bookies数量，Pulsar broker才能向其客户发送确认。实际上可能是：

• （Qa==Qw）或
• （Qa==Qw-1）-->这将通过忽略最慢的bookie来改善延迟。

最终，每一个bookie都必须收到书面通知。但如果我们总是等待所有bookie的回应，我们可能会得到尖峰延迟和不吸引人的尾部延迟。脉冲星毕竟承诺了可预测的延迟。

当Ledger是新主题或发生滚动时，将创建Ledger。滚动是在以下情况下创建新Ledger的概念：

已达到Ledger大小或时间限制

Ledger的所有权（由Pulsar经纪人）发生了变化（稍后将对此进行详细介绍）。

在以下情况下创建片段Fragment：

将创建一个新的Ledger

当当前Fragment ensemble中的Bookie在写入时返回错误或超时时。

当一家bookie无法提供一笔书面交易时，Pulsar broker就会忙于创建一个新的Fragment，并确保该笔交易得到Qw bookies的认可。就像Terminator一样，它不会停止直到消息被持久化。

1. 增加E以优化延迟和吞吐量。以写吞吐量为代价增加冗余的Qw。增加Qa以增加已确认写入的持久性，同时增加额外延迟和更长尾部延迟的风险。

2. E和Qw不是Bookies列表。它们只是表示可以为给定Bookies提供服务的Bookies池有多大。Pulsar在创建新的Ledger或片段时将使用E和Qw。每个碎片的ensemble中都有一组固定的Bookies，这些Bookies永远不会改变。

3. 增加新Bookies并不意味着需要手动重新平衡。自动地，这些新的Bookies将成为新Fragments的候选。加入集群后，创建新的fragments/ledgers后，将立即写入新的Bookies。每个Fragment都可以存储在集群中不同的Bookies子集上！我们不会将主题或Ledgers与给定的Bookies或一组Bookies耦合。

让我们停下来盘点一下。这是一个与Kafka截然不同、更复杂的模型。使用Kafka，每个分区副本都完整地存储在单个代理上。分区副本由一系列段和索引文件组成。这篇博文很好地描述了这一点。

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

回到Pulsar + BookKeeper模式。给定主题的数据分布在多个Bookies中。该主题已被拆分为Ledgers，Ledgers被拆分为Fragments，并通过striping，被拆分为可计算的fragment ensembles子集。当您需要扩展集群时，只需添加更多Bookies，当创建新的片段时，它们就会开始写入。不再需要Kafka式的再平衡。然而，读和写现在不得不在Bookies之间来回跳跃。我们将看到Pulsar是如何管理这一点的，并且会在这篇文章的后面更快。

但现在每个Pulsar BookKeeper都需要跟踪每个主题所包含的Ledgers和fragment。这些元数据存储在ZooKeeper中，如果您丢失了这些元数据，您将面临严重的麻烦。

在存储层中，我们在BookKeeper集群中均匀地编写了一个主题。我们避免了将主题副本耦合到特定节点的陷阱。Kafka的主题就像托布勒龙的棍子，我们的Pulsar主题就像一种气体，膨胀着填满了可用的空间。这避免了痛苦的再平衡。

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

在我的抽象层的第2层，我们有Pulsar broker。Pulsar broker没有不能丢失的持续状态。它们与存储层分离。broker集群本身不执行复制，每个broker只是一个follower跟随者，由一个领导者告诉他该做什么——领导者是一个Pulsar broker。每个主题都由一个Pulsar broker拥有。该代理服务于该主题的所有读写操作。

当Pulsar broker接收到写操作时，它将针对该主题当前Fragment的ensemble执行该写操作。请记住，如果没有发生striping，则每个条目的ensemble与fragment ensemble相同。如果发生striping，则每个条目都有自己的ensemble，它是fragment ensemble的子集。

在正常情况下，当前Ledgers中只有一个fragment。一旦Qa broker确认了写入，Pulsar broker将向生产商客户发送确认。

只有在所有先前的消息也已得到Qa确认的情况下，才能发送确认。如果对于给定的消息，Bookie响应错误或根本没有响应，那么broker将在新的Bookie ensemble（不包括问题Bookie）上创建一个新fragment。

请注意，broker只会等待bookies的Qa acks。

读也要经过所有者。broker作为给定主题的单一入口点，知道已安全地将哪个偏移量保存到BookKeeper手中。它只需要从一家bookies阅读就可以提供阅读服务。我们将在第3层中看到它如何使用缓存服务于内存缓存中的多次读取，而不是将读取发送给BookKeeper。

Pulsar Broker的健康状况由ZooKeeper监控。当代理失败或变得不可用（ZooKeeper）时，所有权发生变化。一个新的Broker成为主题所有者，所有客户机现在都被引导读/写到此新Broker。

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

其工作原理如下：

1. 拥有topic X所有权的当前Pulsar broker（B1）被视为已死亡或不可用（通过ZooKeeper）。

2. 另一个代理（B2）将topic X的当前ledger的状态从OPEN更新为IN_RECOVERY。

3. B2向ledger当前fragment的所有bookie发送fence消息，并等待（Qw Qa）+1响应。收到此数量的响应后，ledger现在被隔离。旧broker（如果它实际上仍然处于活动状态）将无法再进行写入，因为它将无法获得Qa确认（由于隔离异常响应）。

4. B2然后向碎片ensemble中的每个bookie请求他们最后确认的条目是什么。它获取最近的条目id，然后从该点开始向前读取。它确保从该点开始的所有条目（可能之前未向Pulsar broker确认）复制到Qw bookie。一旦B2无法读取和复制任何其他分录，ledger将完全恢复。

5. B2将ledger的状态更改为“已关闭”

6. B2现在可以接受写入并打开新的ledger。

这种架构的伟大之处在于，通过使leader（Pulsar broker）没有状态，split-brain由BookKeeper的fencing功能三位一体地处理。没有split-brain，没有分歧，没有数据丢失。

第2层-Cursor跟踪

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

第3层-Bookie存储

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

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

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

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

读操作首先命中写缓存，因为写缓存具有最新消息。如果缓存未命中，则会命中读缓存。如果第二次缓存未命中，则读缓存在RocksDB中查找请求条目的位置，然后在正确的条目日志文件中读取该条目。它执行预读并更新读缓存，以便后续请求更有可能获得缓存命中。这两层缓存意味着读取通常由内存提供。

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

在后台线程上，写缓存以异步方式对条目日志文件和RocksDB执行大容量写入，这些日志文件和RocksDB通常运行自己的共享磁盘。因此，一个磁盘用于同步写入（日志文件），另一个磁盘用于异步优化写入和所有读取。

在读端，从读缓存或日志条目文件和RocksDB为读卡器提供服务。

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

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

第3层-Pulsar broker缓存

每个主题都有一个充当所有者的broker。所有读写操作都通过该broker进行。这提供了许多好处。

首先，broker可以在内存中缓存日志尾部，这意味着broker可以自己为尾部读取器服务，而不需要BookKeeper。这样就避免了支付网络往返的费用和在bookie上读取磁盘的费用。

broker还知道最后一个添加确认条目的id。它可以跟踪哪条消息是最后一条安全持久化的消息。

当broker在其缓存中没有消息时，它将从该消息fragment ensemble中的bookie请求数据。这意味着tail readers和 catch-up readers之间的reader服务性能差异很大。tail readers可以从Pulsar broker上的内存中读取，而如果写缓存和读缓存都没有数据，则catch-up readers可能需要额外的网络往返和多次磁盘读取的成本。

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

恢复协议

当一个bookie失败时，该bookie上有碎片的所有ledger现在都在复制中。恢复是“重新应用”片段的过程，以确保为每个ledger维护复制因子（Qw）。

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

我们将重点介绍自动恢复模式。

在AutoRecoveryMain过程中，自动恢复可以从一组专用服务器运行，也可以托管在bookie上。其中一个自动恢复过程被选为Auditor。Auditor的职责是发现被击倒的bookie，然后：

从ZK阅读完整的ledger列表，并找到失败的bookie托管的ledger。

对于每个ledger，它将在ZooKeeper中的/复制不足的znode中创建一个重新应用程序任务。

如果Auditor节点失败，则另一个节点将升级为Auditor。Auditor是自动恢复主进程中的一个线程。

AutoRecoveryMain进程还有一个运行复制任务工作线程。每个工作人员都监视/未充分复制的znode以执行任务。

在看到任务时，它会尝试锁定它。如果它无法获得锁，它将进入下一个任务。

如果它确实获得了锁，则：

扫描ledger，查找其本地收受赌注者不是其成员的碎片

对于每个匹配的片段，它将数据从另一个bookie复制到它自己的bookie，用新的集合更新ZooKeeper，并且片段被标记为完全复制。

如果ledger中有剩余未充分复制的fragments，则会释放锁。如果所有fragments都已完全复制，则任务将从中删除/复制不足。

如果fragments没有结束条目id，则复制任务将等待并再次检查，如果fragments仍然没有结束条目id，则在重新应用该fragments之前将隔离ledger。

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

ZooKeeper

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

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

Round Up

• 每个topic主题都有一个所有者broker
• 每个主题在逻辑上被分解为Ledgers, Fragments 和 Entries
• Fragments分布在整个bookie集群中。不存在给定主题与给定bookie的耦合。
• Fragments可以在多个bookie中striped。
• 当Pulsar broker失败时，该代理的主题所有权将转移给另一个broker。Ferning避免了两个可能认为自己是所有者的broker同时向当前主题Ledgers写入内容。
• 当一个bookie失败时，自动恢复（如果启用）将自动向其他bookie“重新应用”数据。如果禁用，则可以启动手动过程
• 代理缓存日志尾部，使它们能够非常高效地为尾部读取器提供服务
• bookie使用日记账为失败提供担保。日志可用于恢复发生故障时尚未写入条目日志文件的数据。
• 所有主题的条目在条目日志文件中交错。查找索引保存在RocksDB中。
• bookie提供如下读取服务：写缓存->读缓存->日志条目文件
• bookie可以通过日志文件、日志条目文件和RocksDB的单独磁盘隔离读写IO。
• ZooKeeper存储Pulsar和BookKeeper的所有元数据。如果ZooKeeper不可用，则Pulsar不可用。
• 存储可以单独扩展到Pulsar代理。如果存储是瓶颈，那么只需添加更多的bookie，他们就可以开始加载，而无需重新平衡。

关于潜在数据丢失的一些初步思考

让我们看看RabbitMQ和Kafka确认的写消息丢失场景，看看它们是否适用于Pulsar。

RabbitMQ使用忽略或自动治疗模式split-brain脑裂。

分区的丢失端丢失自分区开始以来传递的所有未被使用的消息。

Apache Pulsar在理论上不可能在存储层上脑裂。

ApacheKafka，acks=1，带前导副本的broker死亡。

ISR中的跟随者发生故障转移时，可能会丢失消息，因为一旦领导者保留了消息，但可能在跟随者能够获取消息之前发送了ack。

Apache Pulsar没有leader存储节点。如果复制因子（Qw）为2或更多，则单节点故障根本无法导致打开Ledgers的消息丢失（关闭的Ledgers仍可能丢失消息，请参阅下文）。

封闭式Ledgers既暴露在外，也更加丰富。如果Ledgers组的Qa bookie永久丢失，则自动恢复过程无法修复账本。因此，Qa为1是一种危险的设置。开放式Ledgers不是问题，因为broker只需创建一个新片段并负责安全地编写消息。但封闭的Ledgers只能由bookie修复。因此，我们只能保证永久丢失的bookie<Qa时，关闭的Ledgers不会丢失消息。因此，仔细想想你的Qa设置。

让我们考虑一些场景。

情景1（已关闭Ledgers）。E=3，Qw=2，Qa=2。broker将写操作发送给两个bookie。bookie1和bookie2将ack返回给broker，然后broker将ack发送给其客户。一段时间后，Ledgers关闭，新Ledgers启动。现在要产生消息丢失，我们需要bookie1和bookie2都失败。如果任何一家bookie死亡，那么自动恢复协议将生效(注意使用Qa=2。Qa=1可能会使Ledgers无法恢复）。

情景2（未清Ledgers）。E=3，Qw=2，Qa=1。broker将写操作发送给两个bookie。bookie1将ack返回给经纪人，然后broker将ack发送给其客户。bookie2还没有回应。现在，为了产生消息丢失，我们需要broker和bookie1失败，而bookie2没有成功写入。如果只有bookie1死亡，那么broker最终仍然会将消息写入第二个bookie（在一个新的片段中）。

单个节点的故障可能导致消息丢失的唯一方式是Qw为1（这意味着没有冗余）或Qa为1，这意味着自动恢复无法工作。因此，如果要避免消息丢失，请确保有冗余（Qw>=2和Qa>=2）。

Apache Kafka 具有leader分区的节点与ZooKeeper隔离

这导致Kafka的大脑出现短期分裂。

当acks=1时，leader将继续接受写操作，直到它意识到它无法与ZooKeeper对话，此时它将停止接受写操作。与此同时，一个follower被提升为leader。当原始leader成为follower时，在此期间保留给原始leader的任何消息都将丢失。

当acks=all时，如果follower落后并从ISR中移除，则ISR仅由leader组成。然后，leader与ZooKeeper隔离，并在短时间内继续接受acks=所有消息，即使在follower被提升为leader之后也是如此。当leader成为follower时，在短时间内收到的消息将丢失。

Apache Pulsar不能有存储层的分裂大脑。如果当前broker所有者与ZooKeeper隔离，或者遭受长GC，或者其VM被挂起，而另一个broker为所有者，那么仍然只有一个broker可以写入主题。新leader将隔离Ledgers，防止原leader写下任何可能丢失的内容。

Apache Kafka Acks-所有leader故障

follower落在后面，并从ISR中移除。现在，ISR由单个副本组成。即使acks=all，leader也会继续确认消息。leader死了。所有未复制的邮件都将丢失。

ApachePulsar使用了一种基于仲裁的方法，在这种情况下不会发生。只有在Qa bookie将消息持久化到磁盘后，才能发送ack。

Apache kafka-同时断电（数据中心断电）

Kafka将确认一条写入内存的消息。它定期同步到磁盘。当数据中心断电时，所有服务器都可能同时脱机。消息可能仅在所有副本的内存中。这条信息现在丢失了。

Apache Pulsar仅在Qa bookie确认消息后确认消息。只有当一个条目被保存到磁盘上的日志文件中时，bookie才会确认该条目。除非同时发生多个磁盘故障，否则所有服务器同时断电不应丢失消息。

到目前为止，Apache Pulsar看起来相当健壮。我们得看看在混乱测试中的表现如何。

结论

还有更多细节我要么错过了，要么还不知道。ApachePulsar在协议和存储模型方面比ApacheKafka复杂得多。

Pulsar集群的两个突出特征是：

• 将代理与存储分离，再加上BookKeepers fencing功能，优雅地避免了可能导致数据丢失的大脑分裂场景。
• 将主题分解成ledgers 和 fragments，并将它们分布在一个Pulsar集群中，可以让Pulsar集群轻松扩展。新数据自动开始写入新的bookies。不需要再平衡。

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

我觉得Pulsar和BookKeeper是下一代数据流系统的一部分。他们的协议经过深思熟虑，相当优雅。但随着复杂性的增加，bug的风险也随之增加。

原文地址：https://jack-vanlightly.com/blog/2018/10/2/understanding-how-apache-pulsar-works