【翻译】理解pulsar如何工作

[toc]

译者注：

翻译：pulsar broker：pulsar经纪人（代理）。Ledger：分类账。fragment：片段。cursor：游标。Entity：条目。split-brain：裂脑（不是很好的翻译）。chaos testing：混乱测试

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

消息系统

我将撰写一系列关于Apache Pulsar的博客文章，包括一些Kafka vs Pulsar的帖子。首先，我将在Pulsar集群上运行一些极端情况测试（原文为：chaos tests，译者注），就像我使用RabbitMQ和Kafka一样，看看它有什么故障模式及其消息丢失情况。

我将尝试通过利用管理员或开发人员的设计缺陷，实现错误或不良配置来做到这一点。

在这篇文章中，我们将介绍Apache Pulsar设计，以便我们可以更好地设计故障情况。这篇文章不适合那些想要了解如何使用Apache Pulsar，相反，这篇文章适合想要了解它如何工作的人。我一直在努力以简单易懂的方式对其架构进行清晰的概述。我很感激有关此文章的任何反馈。

申明

我感兴趣的主要声明（指Apache Pulsar的申明，译者注）是：

保证没有消息丢失（如果应用推荐的配置，并且您的整个数据中心没有出故障）
强大的实时订阅保证
可预测的读写延时

Apache Pulsar选择一致性而不是可用性，就像它的姐妹项目BookKeeper和ZooKeeper一样。我们尽一切努力保持一致性。

我们将看看Pulsar的设计，看看这些说法是否有效。在下一篇文章中，我们将对该设计的实现进行测试。我不会在这篇文章中讨论地理复制，我们将在以后看一下，现在，我们只关注一个集群。

多层抽象

Apache Pulsar具有主题和订阅的高级概念，其最低级别的数据存储在二进制文件中，这些文件交叉分布在多个服务器上的多个主题的数据。在它们之间是无数的细节和活动部分。我个人觉得如果把它分成不同的抽象层，就更容易理解Pulsar架构，所以这就是我在这篇文章中要做的。

让我们先看一下分层：

第一层-主题，订阅和游标

这不是关于可以使用Apache Pulsar构建的消息传递体系结构的帖子。我们将简要介绍主题，订阅和游标的基础知识，但不是关于Pulsar支持的更广泛的消息传递模式的深度。

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

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

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

一个主题可以有多个附加订阅。订阅不包含数据，仅包含元数据和游标。
Pulsar通过允许消费者将Pulsar主题视为在消费者确认后删除消息的队列，或者像消费者可以根据需要回放游标的日志来提供排队和日志语义。存储模型底层是相同的 - 日志。

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

但是，如果存在涵盖主题的数据保留策略，则一旦通过策略边界（主题的大小，主题中的时间），就会删除消息。

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

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

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

最后，有一些类似于Kafka主题的分区主题。区别在于Pulsar中的分区也是主题。就像kafka一样，生产者可以使用散列算法或明确选择分区，循环发送消息。

这是对高级概念的旋风式介绍，我们现在将深入研究。请记住，这不是从顶层来学习使用Apache Pulsar的入门读物，而是看看它在底层的工作原理。

第二层-逻辑存储模型

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

首先，BookKeeper在一组节点上存储数据。每个BookKeeper节点都称为Bookie。其次，Pulsar和BookKeeper都使用Apache Zookeeper来存储元数据和监控节点健康状况。

一个主题实际上是一个Ledgers流。 Ledger本身就是一个日志。因此，我们将一系列子日志（Ledgers）中组合成父日志（主题）。

分类帐（Ledgers）附加到主题，条目（消息或消息组）附加到分类帐。 Ledgers一旦关闭，是不可改变的。分类帐作为一个单元被删除，也就是说，我们不能删除单个条目而是删除整个分类帐。

Ledgers本身也被分解成片段。片段是BookKeeper集群中最小的分布单元（根据您的观点，条带化可能会使该声明无效（原文为：depending on your perspective, striping might invalidate that claim，译者注））。

主题是Pulsar概念。分类帐，片段和条目是BookKeeper的概念，尽管Pulsar理解并使用了分类帐和条目。

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

每个Ledger有三个关键配置：

集合大小（Ensemble Size (E)）
写法定大小（Write Quorum Size (Qw)）
确认法定大小（Ack Quorum Size (Qa)）

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

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

通过将Bookies标记为属于特定机架，还可以选择配置机架感知。机架可以是逻辑构造（例如：云环境中的可用区域）。通过机架感知策略，Pulsar客户端的BookKeeper客户端将尝试从不同的机架中选择Bookies。也可以插入自定义策略以执行不同类型的选择。

集合大小（E）控制Pulsar写入的Ledger可用的Bookies池的大小。每个片段可能有不同的集合，经纪人（pulsar broker）将在创建片段时选择一组Bookies（broker是选取一组bookies作为存储体，这一组存储体就存储这个片段（fragment），译者注），但整体将始终是由E指示的大小。必须有足够的Bookies可用于覆盖E。

Write Quorum（Qw）是Pulsar写入条目的实际Bookies数。它可以等于或小于E。A fragment of 8 entries stored across an ensemble of 3 with each entry written to 3 bookies.

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

Ack Quorum（Qa）是必须承认写入的Bookies的数量，Pulsar经纪人将其确认发送给其客户（客户端：译者注）。在实践中它可能是：

(Qa == Qw) or
(Qa == Qw -1) —> This will improve latency by ignoring the slowest bookie.

最终，每个预订者都必须收到写确认。但是，如果我们总是等待所有的Bookies做出回应，我们就会得到波动的延迟（原文为：spiky latency意为高低不平的延时，且作者有拼写错误，译者注）和没有吸引力的尾部延迟。 Pulsar毕竟承诺可预测的延迟。

（译者注：topic被分为多个Ledger分类账，而一个分类账需要多个bookie来做记录（高可读和容灾），这多个bookie就为了方便描述就被描述为fragment，但是ledger和fragment并不等价，一个Ledger有多个fragment。topic和Ledger的关系有点像fragment和bookie的关系）

当创建新主题或发生翻转（Roll-over）时，会创建分类帐。翻转是在以下任何一种情况下创建新Ledger的概念：

一个分类账的大小或者时间限制已经满足
一个分类账所有权（一个pulsar经纪人）发生改变

当一下情形发生时，片段被创建：

一个分类账被创建
当前片段集合的一个bookie在写操作时返回一个错误，或者超时

当一个bookie无法提供写作时，Pulsar经纪人就会忙着创建一个新的片段，并确保写入得到Qw个bookie的认可。就像终结者一样，它不会停止，直到该消息被持久化。

Insight＃1：增加E以优化延迟和吞吐量。以写入吞吐量为代价增加Qw以实现冗余。增加Qa，就增加了已确认写入的持久性，也增加了额外延迟和更长尾部延迟的风险。

Insight＃2：E和Qw不是Bookies列表。它们只是表明可以为给定的Ledger服务的Bookies池有多大。 Pulsar将在创建新的Ledger或Fragment时使用E和Qw。每个片段在其整体中都有一组固定的Bookies，永远不会改变。

Insight＃3：添加新Bookies并不意味着需要执行手动重新平衡。这些新的Bookies将自动成为新片段的候选者。加入群集后，将在创建新的片段/分类帐后立即写入新的Bookies。每个片段都可以存储在群集中不同的Bookies子集中！我们不会将主题或分类帐连接到给定的Bookie或Bookies集合。

让我们停下来评估一下。对于卡夫卡来说，这是一个非常不同且更复杂的模型。使用Kafka，每个分区副本完全存储在单个代理上。分区副本由一系列段和索引文件组成。这篇博文（ https://thehoard.blog/how-kafkas-storage-internals-work-3a29b02e026）很好地描述了它。

Kafka模型的优点在于它简单快速。所有读写都是顺序的。糟糕的是，单个代理必须有足够的存储空间来处理该副本，因此非常大的副本可能会迫使您拥有非常大的磁盘。第二个缺点是，在扩展集群时重新平衡分区变得必要。这可能是痛苦的，需要良好的计划和执行才能在没有任何障碍的情况下拉开。

回到Pulsar + BookKeeper模型。给定主题的数据分布在多个Bookies中。该主题已分为Ledgers，而Ledgers分为片段，并带有条带化，成为可计算的片段集合子集。当您需要扩展集群时，只需添加更多Bookies，它们就会在创建新片段时开始写入。不再需要卡夫卡式的再平衡。但是，读取和写入现在必须在Bookies之间跳跃一点。我们将在这篇文章看到Pulsar如何管理并高速处理。

但现在每个Pulsar经纪人都需要跟踪每个主题所包含的Ledgers和Fragments。这个元数据存储在ZooKeeper中，如果你丢失了，那么你就会陷入严重的困境。

在存储层中，我们编写的主题均匀的分布到BookKeeper集群。我们避免了将Topic副本耦合到特定节点的陷阱。卡夫卡主题就像Toblerone（一种巧克力冰激凌，用中文可以说为老冰棍的棍子，译者注）的棒子一样，我们的Pulsar主题就像一个气体膨胀来填补可用空间。这避免了痛苦的再平衡。

第二层-pulsar的broker和topic所有权

同样在我的抽象层的第2层，我们有Pulsar Brokers。 Pulsar Brokers（经纪人）没有会丢失的持久状态。它们与存储层分开。 BookKeeper集群本身并不执行复制，每个Bookie只是一个跟随者，被领导者告知应该做什么 - 领导者是Pulsar经纪人。 每个主题都只由单独一个Pulsar经纪人拥有。该经纪服务于该主题的所有读写操作。

当Pulsar经纪人接收到写入时，它将针对该主题的当前片段的集合执行该写入。请记住，如果没有条带化，则每个条目（Entity，条目可以认为是一条记录，类似sql中的一行记录，译者注）的集合与片段集合相同。如果发生条带化，那么每个条目都有自己的集合，这是片段集合的一个子集。

在一般情况下，当前的Ledger中将有一个Fragment。一旦Qa个经纪人承认写入，Pulsar经纪人将向生产者客户发送确认。

只有在所有先前消息都已通过Qa个确认时，才能发送确认。如果对于给定的消息，Bookie响应错误或根本没有响应，则经纪人将在新的Bookies集合上创建新的片段（不包括问题Bookie）。

对于特定的主题，一个经纪人服务所有的读写

请注意，经纪人只会等待来自bookie的Qa ack。

读取也通过所有者。作为给定主题的单一入口点的经纪人知道哪些偏移已经安全地保存到BookKeeper。它只需要从一个Bookie读取即可进行读取。我们将在第3层中看到它如何使用缓存从其内存缓存中提供许多读取，而不是将读取发送到BookKeeper。

只需要从一个bookie去读取

Pulsar Broker的健康状况由ZooKeeper监控。当代理失败或变得不可用（对ZooKeeper）时，会发生所有权更改。新的代理成为主题所有者，然后所有客户端都被定向读取/写入此新代理。

BookKeeper有一个非常重要的功能，称为Fencing。 Fencing允许BookKeeper保证只有一个写入者（Pulsar经纪人）可以写入分类账。

它按如下流程工作

拥有主题X的当前Pulsar经纪人（B1）被视为已死或不可用（通过ZooKeeper来监控并判定）。
另一个代理（B2）将主题X的当前分类帐的状态从OPEN更新为IN_RECOVERY。
B2向分类账的当前片段的所有bookies发送围栏消息，并等待（Qw-Qa）+1响应。收到此响应数后，分类帐现在会被围起来。旧代理如果它实际上仍然存活，则无法进行写入，因为它无法获得Qa确认（由于屏蔽异常响应）。
B2然后从片段集合中的每个bookie请求他们最后确认的条目是什么。它需要最新的条目ID，然后从该点开始向前阅读。它确保从那一点开始的所有条目（可能以前未向Pulsar经纪人确认）都会被复制到Qw个bookies。一旦B2无法读取并复制任何更多条目，分类帐将完全恢复。
B2将分类账的状态改变为closed
B2现在可以在新的分类账上接受写和读操作

关于这种架构的伟大之处在于，通过让领导者（Pulsar经纪人）没有状态，BookKeeper的围栏功能可以很好地处理裂脑（原文为split-brain，译者注）问题。没有裂脑，没有分歧，没有数据丢失。

第二层-游标追踪

每个订阅都存储一个游标。游标是日志中的当前偏移量。订阅将其游标存储在BookKeeper的分类帐中。这使游标跟踪可以像主题一样进行扩展。

第三层-bookie存储

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

write

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

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

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

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

BookKeeper允许您将磁盘IO与读写隔离。写入都按顺序写入日志文件，可以存储在专用磁盘上，并以组的形式提交，以获得更高的吞吐量。之后，从写入者的角度来看，不需要磁盘IO的同步。数据只写入内存缓冲区。

写缓存在后台线程上异步执行批量写入Entry Log文件和RocksDB，所以写缓存通常运行自己的共享磁盘。因此，一个磁盘用于同步写入（日志文件），另一个磁盘用于异步优化写入和所有读取。（译者注：一个磁盘用于批量写入，另一个磁盘接受写入和读取请求（这些都是不连续的））

read

在读的方面，读者要么被读缓存服务，要么被日志条目文件和RocksDB服务

summary

还要考虑到写入可以使入口网络带宽（ ingress network bandwidth）饱和，并且读取可以使出口网络带宽（egress network bandwidth）饱和，但它们不会相互影响。

这种优雅的隔离读取来自磁盘和网络级别的写入。

第三层-pulsar代理人缓存

每个主题都仅有一个代理所有者的代理。所有读写都通过该代理进行。这提供了许多好处。

首先，代理可以将日志尾部缓存在内存中，这意味着代理可以在不需要BookKeeper的情况下为尾部读取器（tailing readers，译者注）提供服务。这避免了支付网络往返（network round-trip，译者注）的费用以及Bookie上可能的磁盘读取。

经纪人也知道Last Add Confirmed条目的id。它可以跟踪哪条消息是最后一个安全持久的消息。

当代理在其缓存中没有消息时，它将从该消息的片段集合中的一个Bookie请求数据。这意味着尾部读取器和追赶读取器（catch-up readers，译者注）之间的读取服务性能差异很大。尾部读取器可以从Pulsar代理的内存中提供，而如果写入和读取高速缓存都没有数据，则追赶读取器可能必须承担额外的网络往返和多次磁盘读取的成本。

因此，我们从高层次上涵盖了消息的逻辑和物理表示，以及Pulsar集群中的不同参与者及其相互之间的关系。有很多细节尚未涵盖，但我们会将其作为以后的练习。

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

恢复协议

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

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

我们将聚焦自动回复机制

自动恢复可以在AutoRecoveryMain流程中从一组专用服务器运行，也可以在Bookies上托管。其中一个自动恢复进程被选为审计员（原文为：Auditor，译者注）。审计员的作用是检测崩溃的Bookies然后：

阅读ZK的完整分类帐清单，找到托管（原文为：hosted on）在失败的Bookies上的分类帐。
对于每个分类帐，它将在ZooKeeper中的/underreplicated znode中创建重新复制任务。

如果Auditor节点出现故障，则另一个节点将升级为Auditor。 Auditor是AutoRecoveryMain过程中的一个线程。

AutoRecoveryMain进程还有一个运行Replication Task Worker的线程。每个工作人员（worker）都会监视未充分复制的znode以查找任务。

在看到任务时，它会尝试锁定它（获取锁，译者注）。如果它无法获取锁定，它将进入下一个任务。

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

扫描分类帐，查找本地bookie不属于的碎片
对于每个匹配的片段，它将来自另一个bookie的数据复制到自己的bookie，使用新的集合更新ZooKeeper，并将片段标记为完全复制。

如果分类帐仍然存在未复制的碎片，则释放锁定。如果所有片段都完全复制，则从 /underreplicated 删除任务。

如果片段没有结束条目ID，则复制任务将等待并再次检查，如果片段仍然没有结束条目ID，则它会在重新复制片段之前对分类帐进行隔离（fences）。

因此，使用自动恢复模式，Pulsar集群能够完全复制细节，以确保每个分类帐的正确复制因子。管理员只需确保部署适量的bookies。

ZooKeeper

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

这意味着如果ZooKeeper发生故障，一切都变得不可用，并且所有Pulsar节点缓存都将被擦除。因此，在恢复服务时，理论上可能存在由于所有读取都到达BookKeeper而导致的延迟峰值。

流程

主题有一个所有者经纪人
每一个主题逻辑上被分为，分类账们，片段们和条目们
片段分布在bookie集群中。给定的主题与给定的bookies没有耦合。
片段可以跨多个bookies条带化
当Pulsar经纪人失败（可以翻译为崩溃，但译者不愿意：译者注）时，该经纪人的主题的所有权将故障转移给另一个经纪人。 Fencing避免了两个可能认为自己的所有者同时实际写入当前主题分类帐的经纪人。
当bookie失败时，自动恢复（如果启用）将自动将数据“重新复制”到其他bookies。如果禁用，则可以启动手动过程。
经纪人缓存日志尾部，使他们能够非常有效地为尾部读取器提供服务
bookie使用日志来提供失败保证。该日志可用于恢复发生故障时，尚未写入Entry Log文件的数据。
所有主题的条目都在Entry Log文件中交错。查找索引保存在RocksDB中。
Bookies按如下方式提供服务: Write Cache -> Read Cache -> Log Entry files
Bookies可以通过单独的磁盘隔离读取写入IO，用于日志文件，日志条目文件和RocksDB
ZooKeeper存储Pulsar和BookKeeper的所有元数据。如果ZooKeeper不可用，则Pulsar不可用
存储可以单独扩展到Pulsar经纪人。如果存储是瓶颈，那么只需添加更多的bookies ，他们将开始承担负载而无需重新平衡。

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

–这一部分未完全翻译

让我们来看看RabbitMQ和Kafka承认的写消息丢失情况，看看它们是否适用于Pulsar。

RabbitMQ具有Ignore或Autoheal模式的裂脑

分区的丢失方丢失了自分区开始以来未传递的任何消息。

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

结论

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

pulsar集群的两个显著特点是：

将存储器与存储分离，结合BookKeepers防护功能，可以优雅地避免可能引起数据丢失的裂脑情况。
将主题分解为分类帐和碎片，并在群集中分发这些主题，使Pulsar群集可以轻松扩展。新数据自动开始写入新的bookie。不需要重新平衡。

此外，我甚至没有进行到地理复制和分层存储，这也是令人惊叹的功能。

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

标签：Apache Pulsar