在上一篇文章中，我们检查了RabbitMQ集群的容错能力和高可用性。 现在让我们深入研究Apache Kafka。

在这里，复制单元是一个分区。 每个主题都有一个或多个部分。 每个部分都有带或不带跟随者的领导者。 创建主题时，将指示分区数和复制率。 通常的值为3，表示三句话：一位领导者和两位跟随者。



图 1.在三个经纪人中分配四个部分

所有的读写请求都发送给领导者。 关注者会定期向领导发送请求，以接收最新消息。 消费者从不求助于追随者，追随者仅出于冗余和容错的目的而存在。

部分失败

当经纪人崩溃时，几个部门的领导者常常会失败。 在每个节点中，来自另一个节点的跟随者都将成为领导者。 实际上，情况并非总是如此，因为同步因素也会影响：是否有同步的跟随者，如果没有，则允许过渡到非同步副本。 但是现在，我们不要使其复杂化。

经纪人3离开网络-对于第2部分，选举了经纪人2的新领导者。


图 2.经纪人3去世，其经纪人2的跟随者当选为第2节的新负责人

然后经纪人1离开，第1部分也失去了其领导者，而其领导者则由经纪人2承担。


图 3.仅剩一个经纪人。 所有领导者都在同一个零冗余代理人上。

当代理1返回到网络时，他添加了四个关注者，从而为每个部分提供了一些冗余。 但是所有领导者仍然留在经纪人2上。


图 4.领导人留在经纪人2上

当代理3上升时，我们将每个部分返回三个副本。 但所有领导人仍在经纪人2上。


图 5.恢复经纪人1和3后领导者的不平衡分配

与RabbitMQ相比，Kafka具有更好地平衡领导者的工具。 在那里，您必须使用第三方插件或脚本，这些插件或脚本通过减少迁移过程中的冗余来更改用于迁移主节点的策略。 另外，对于大队列，必须在同步期间忍受不可访问性。

卡夫卡在领导角色上拥有“首选线索”的概念。 创建主题部分后，Kafka会尝试将领导者均匀地分布在节点上，并将这些第一个领导者标记为首选。 随着时间的流逝，由于服务器重新引导，故障和连接故障，领导者可能最终会出现在其他节点上，如上文所述的极端情况。

为了解决这个问题，Kafka提供了两种选择：

• auto.leader.rebalance.enable = true选项允许控制器节点自动将领导者重新分配回首选副本，从而恢复统一分发。
  
• 管理员可以运行kafka-preferred-replica-election.sh脚本来手动重新分配。

图 6.重新平衡后的副本

这是故障的简化版本，但实际情况更为复杂，尽管这里没有什么太复杂的。 全部归结为同步副本（同步副本，ISR）。

同步副本（ISR）

ISR是被视为“同步”（同步）的分区的一组副本。 有领导者，但可能没有跟随者。 如果追随者在复制副本.lag.time.max.ms间隔到期之前对所有领导者消息进行了精确复制，则认为该追随者已同步。

如果满足以下条件，则从ISR集中删除该关注者：

• 没有请求间隔replica.lag.time.max.ms的采样请求（考虑为无效 ）
  
• 没有时间更新时间间隔的副本copy.lag.time.max.ms （被认为很慢）

追随者会在interval.fetch.wait.max.ms间隔（默认为500毫秒）中进行获取请求。

为了清楚地说明ISR的目的，您需要查看生产者（生产者）的确认和一些失败情况。 生产者可以选择经纪人发送确认的时间：

• acks = 0，不发送确认
  
• acks = 1，在领导者将消息写入其本地日志后发送确认
  
• acks =全部，ISR中的所有副本均已将消息写入本地日志后发送确认

用Kafka术语，如果ISR已保存该消息，则该消息“已提交”。 Acks = all是最安全的选择，但也会带来额外的延迟。 让我们看一下两个故障示例，以及不同的“ acks”选项如何与ISR概念相互作用。

Acks = 1和ISR

在此示例中，我们将看到，如果领导者不等待保存所有关注者的每条消息，那么如果领导者失败，则数据可能会丢失。 通过设置unclean.leader.election.enable可以启用或禁用转到不同步的跟随器。

在此示例中，制造商设置为acks = 1。 该部分分布在所有三个代理中。 经纪人3落后了，它在八秒钟前与领导者同步，现在落后了7456条消息。 经纪人1仅落后一秒钟。 我们的生产者发送消息并迅速收到确认，而对于领导者所不希望的缓慢或死忠的追随者却没有开销。


图 7.具有三个副本的ISR

代理2失败，并且制造商收到连接错误。 从领导层过渡到经纪人1后，我们丢失了123条消息。 经纪人1的追随者是ISR的一部分，但在领导者跌倒时并未与他完全同步。


图 8.失败后，消息将丢失

在bootstrap.servers配置中，制造商列出了几个经纪人，他可以询问另一个经纪人，谁成为了该部分的新负责人。 然后，他与代理1建立连接并继续发送消息。


图 9.短暂休息后恢复发送消息

经纪人3进一步落后。 它发出获取请求，但无法同步。 这可能是由于代理之间的网络连接速度慢，存储问题等。已将其从ISR中删除。 现在，ISR包含一个副本-领导者！ 制造商继续发送消息并收到确认。


图 10.经纪人3的关注者已从ISR中删除

代理1倒下，领导者的角色转给代理3，丢失15286条消息！ 制造商收到连接错误消息。 仅由于设置unclean.leader.election.enable = true ，才有可能在ISR之外进入领导者。 如果将其设置为false ，那么将不会发生转换，并且所有读取和写入请求都将被拒绝。 在这种情况下，我们正在等待代理1的返回，其中代理1的原始数据在副本中，它将再次带头。


图 11.经纪人1滴。 如果发生故障，则会丢失大量消息

制造商与最后一个经纪人建立了联系，并确定他现在是该部门的负责人。 他开始向经纪人3发送消息。


图 12.短暂休息后，消息再次发送到第0部分

我们看到，除了短暂的中断以建立新的连接并寻找新的领导者之外，制造商还不断发送消息。 此配置通过一致性（数据安全性）提供可访问性。 卡夫卡丢失了数千封邮件，但继续接受新邮件。

Acks =全部和ISR

让我们再次重复这种情况，但是用acks = all 。 将代理3延迟平均四秒钟。 制造商发送的消息带有acks = all ，现在没有收到快速响应。 领导者等待，直到ISR中的所有消息都存储了该消息。


图 13.具有三个副本的ISR。 一个很慢，导致录制延迟

在四秒钟的额外延迟之后，代理2发送ack。 现在，所有副本均已完全更新。


图 14.所有副本均保存消息并发送确认

经纪人3现在更加落后，并已从ISR中删除。 由于ISR中没有慢速副本，因此可以大大减少延迟。 代理2现在仅等待代理1，他的平均延迟为500毫秒。


图 15.代理3上的副本已从ISR中删除

然后经纪人2倒下，领导权传递给经纪人1而不会丢失任何消息。


图 16.经纪人2下跌

制造商找到新的领导者并开始向他发送消息。 由于现在ISR由一个副本组成，因此延迟仍然减少了！ 因此， acks = all选项不会增加冗余。


图 17.代理1上的副本处于领先地位，而不会丢失消息

然后经纪人1倒下，领导权传递给经纪人3，损失了14,238条消息！


图 18. Broker 1死了，设置不整洁的领导层过渡导致大量数据丢失

我们无法将unclean.leader.election.enable选项设置为true 。 默认情况下，它为false 。 设置acks =全部为unclean.leader.election.enable = true可提供可访问性以及一些其他数据安全性。 但是，正如您所看到的，我们仍然会丢失消息。

但是，如果我们想提高数据安全性怎么办？ 您可以设置unclean.leader.election.enable = false ，但这并不一定能防止数据丢失。 如果领导者摔倒并随身携带了数据，那么消息仍然会丢失，并且可访问性也会丢失，直到管理员恢复情况为止。

最好保证所有消息的冗余，否则拒绝记录。 然后，至少从代理的角度来看，只有同时发生两个或多个故障，才有可能发生数据丢失。

Acks =全部，最小不同步副本和ISR

使用min.insync.replicas主题配置，我们可以提高数据安全性。 让我们再次遍历最后一个场景的最后一部分，但是这次是min.insync.replicas = 2 。

因此，代理2具有副本领导者，并且代理3的跟随者已从ISR中删除。


图 19.两个副本的ISR

经纪人2倒下，领导权传递给经纪人1而不会丢失任何消息。 但是现在，ISR仅包含一个副本。 这与接收记录的最小数量不对应，因此代理对记录尝试进行响应，并显示NotEnoughReplicas错误。


图 20. ISR的数量比min.insync.replicas中指定的数量少一

此配置会牺牲可用性以保持一致性。 在确认消息之前，我们保证将其记录在至少两个副本上。 这使制造商更有信心。 在此，只有两个副本在短时间内同时失败，消息丢失才有可能，直到将消息复制到其他跟随者为止，这是不可能的。 但是，如果您是超级偏执狂 ，则可以将复制比率设置为5，将min.insync.replicas设置为3。然后，必须同时跌倒三个代理才能丢失记录！ 当然，为了获得这种可靠性，您将付出额外的延迟。

需要数据安全性的可访问性时

与RabbitMQ一样 ，有时为了数据安全性，必须具有可访问性。 您需要考虑一下：

• 发布者可以返回错误，而更高的服务或用户以后再试吗？
  
• 发布者可以在本地或数据库中保存邮件以供稍后重试吗？

如果答案是否定的，那么优化可访问性将提高数据安全性。 如果选择可用性而不是放弃录制，则丢失的数据将更少。 因此，一切都取决于找到平衡，而决定取决于具体情况。

ISR的含义

ISR套件使您可以在数据安全性和延迟之间选择最佳的平衡。 例如，要确保在发生故障时可以访问大多数副本，就延迟而言，将死副本或慢副本的影响降到最低。

我们根据自己的需要选择copy.lag.time.max.ms的值。 本质上，此参数表示我们准备接受acks = all的延迟时间。 默认值为十秒。 如果这对您来说太长了，您可以减少它。 然后，ISR中更改的频率将增加，因为追随者将被更频繁地删除和添加。

RabbitMQ只是需要复制的一组镜像。 慢速镜像会带来额外的延迟，并且在检查每个节点的可用性的数据包到期之前（网络滴答声），可以预期死镜像的响应。 ISR是一种避免延迟增加的问题的有趣方法。 但是我们有失去冗余的风险，因为ISR只能简化为领导者。 为避免这种风险，请使用min.insync.replicas设置。

客户连接保证

在制造商和消费者的bootstrap.servers设置中，可以指定几个代理来连接客户端。 这个想法是，当您断开一个节点的连接时，客户端可以使用几个备用节点来打开连接。 这些不一定是部门负责人，而仅仅是引导的跳板。 客户端可以询问他们读/写部分的负责人位于哪个节点上。

在RabbitMQ中，客户端可以连接到任何主机，并且内部路由会在必要时发送请求。 这意味着您可以在RabbitMQ的前面安装负载均衡器。 Kafka要求客户端连接到托管相应分区负责人的主机。 在这种情况下，负载均衡器将无法交付。 bootstrap.servers列表至关重要，因此客户端可以访问正确的节点并在崩溃后找到它们。

卡夫卡共识架构

到目前为止，我们还没有考虑集群如何发现经纪人的倒台以及如何选择新的领导者。 要了解Kafka如何与网络分区一起使用，您首先需要了解共识架构。

每个Kafka群集都与Zookeeper群集一起部署-这是一种分布式共识服务，它允许系统在某些给定状态下以一致性优先于可用性来达成共识。 批准读写操作需要大多数Zookeeper节点的同意。

Zookeeper存储集群状态：

• 主题，部分，配置，当前领导者副本，首选副本的列表。
  
• 集群成员。 每个代理将ping到Zookeeper群集中。 如果他在给定的时间内没有收到ping命令，则Zookeeper将使该代理无法访问。
  
• 控制器的主节点和辅助节点的选择。

控制器节点是负责选举副本领导者的Kafka经纪人之一。 Zookeeper向控制器发送集群成员资格和主题更改的通知，并且控制器必须根据这些更改采取行动。

例如，以一个具有十个部分且复制系数为3的新主题为例。控制器必须选择每个部分的领导者，以尝试在代理之间最佳地分布领导者。

对于每个部分，控制器：

• 更新Zookeeper中有关ISR和领导者的信息；
  
• 向发布此部分副本的每个经纪人发送一个LeaderAndISRCommand命令，以通知经纪人有关ISR和领导者的信息。

当具有领导者的经纪人跌倒时，Zookeeper将通知发送给控制者，然后他选择新的领导者。 再次，控制器首先更新Zookeeper，然后向每个代理发送命令，以通知他们领导层的变化。

每个负责人负责招募情监侦。 copy.lag.time.max.ms设置确定谁将去那里。 当ISR发生变化时，领导者会将新信息传递给Zookeeper。

Zookeeper始终会收到有关任何更改的通知，以便在发生故障时，管理层可以平稳地转移到新的主管。


图 21.卡夫卡共识

复制协议

了解复制详细信息有助于您更好地了解潜在的数据丢失情况。

样品要求，原木末端偏移量（LEO）和高水位线（HW）

我们认为追随者会定期将获取请求发送给领导者。 默认间隔为500毫秒。 这与RabbitMQ的不同之处在于，在RabbitMQ中，复制不是由队列镜像启动，而是由向导启动。 主机将更改推送到镜像。

领导者和所有跟随者保留对数末端偏移（LEO）和高水位（HW）标签。 LEO标记将最后一条消息的偏移量存储在本地副本中，而硬件将最后一次提交的偏移量存储在本地副本中。 请记住，对于提交状态，该消息必须保存在所有ISR副本中。 这意味着LEO通常比HW稍微领先。

领导者收到消息后，将其保存在本地。 跟随者发出获取请求，并传递他的LEO。 然后，领导者发送一个从此LEO开始的消息包，并发送当前的硬件。 领导者收到有关所有副本已按给定偏移量保存邮件的信息时，他将移动HW标记。 只有领导者才能移动硬件，因此所有关注者都将在对其请求的响应中知道当前值。 这意味着追随者在硬件的报告和知识方面都可能落后于领导者。 消费者仅收到当前硬件之前的消息。

请注意，“持久”是指写入内存，而不是磁盘。 为了提高性能，Kafka以指定的时间间隔同步到磁盘。 RabbitMQ也有这样的间隔，但是只有在主服务器和所有镜像将消息写入磁盘后，它才会向发布者发送确认。 由于性能原因，Kafka开发人员决定在消息写入内存后立即发送ack。 Kafka依靠这样的事实，即冗余可弥补仅在内存中短期存储已确认消息的风险。

领导失败

当领导者跌倒时，Zookeeper会通知控制器，然后他选择一个新的领导者副本。 新领导人根据他的LEO设置了新的硬件标记。 然后关注者会收到有关新领导者的信息。 根据Kafka的版本，关注者将选择以下两种情况之一：

1. 将本地日志截断到著名的硬件，并在此标记之后向新领导发送消息。
   
2. , HW , . , .

:

• , ISR, Zookeeper, . ISR, «», . , . Kafka , . , , HW . , acks=all .
  
• . , . , , , , , .

c

, : HW ( ). , RabbitMQ . . , « ». . .

Kafka — , , RabbitMQ, . . Kafka — , . . Kafka HW ( ) , . , , , LEO.

ISR . , , , ISR. .

Kafka , RabbitMQ, , . Kafka , .

:

• 1. , Zookeeper.
  
• 2. , Zookeeper.
  
• 3. , Zookeeper.
  
• 4. , Zookeeper.
  
• 5. Kafka, Zookeeper.
  
• 6. Kafka, Zookeeper.
  
• 7. Kafka Kafka.
  
• 8. Kafka Zookeeper.

.

1. , Zookeeper

. 22. 1. ISR

3 1 2, Zookeeper. 3 . replica.lag.time.max.ms ISR . , ISR, . Zookeeper , .


. 23. 1. ISR, replica.lag.time.max.ms

(split-brain) , RabbitMQ. .

2. , Zookeeper

. 24. 2.

, Zookeeper. , ISR , , . , . , . Zookeeper , .


. 25. 2. ISR

3. , Zookeeper

Zookeeper, . ISR. Zookeeper , , .


. 26. 3.

4. , Zookeeper

. 27. 4.

Zookeeper, .


. 28. 4. Zookeeper

Zookeeper . . , acks=1 . , ISR . Zookeeper, , .

acks=all , ISR , . ISR, - .

. , , , HW, , . . , . , , .


. 29. 4. 1

5. Kafka, Zookeeper

Kafka, Zookeeper. ISR, , .


. 30. 5. ISR

6. Kafka, Zookeeper

. 31. 6.

, Zookeeper. acks=1 .


. 32. 6. Kafka Zookeeper

replica.lag.time.max.ms , ISR , , Zookeeper, .

, Zookeeper , .


. 33. 6.

, . 60 . .


. 34. 6.

, . , Zookeeper , . HW .


. 35. 6.

, acks=1 min.insync.replicas 1. , , , , — , . , acks=1 .

, , ISR . - . , , acks=all , ISR . . — min.insync.replicas = 2 .

7. Kafka Kafka

, Kafka . , 6. .

8. Kafka Zookeeper

Zookeeper Kafka. , Zookeeper, . , , , Kafka.

, , , . , , , .

- Zookeeper, acks=1 . Zookeeper . acks=all .

min.insync.replicas , , 6.

, Kafka:

• , acks=1
  
• (unclean) , ISR, acks=all
  
• Zookeeper, acks=1
  
• , ISR . , acks=all . , min.insync.replicas=1 .
  
• . , . .

, , . — acks=all min.insync.replicas 1.

RabbitMQ Kafka

. RabbitMQ . , . RabbitMQ. , . . , ( ) .

Kafka . . . , . , , . , - , . , .

RabbitMQ Kafka . , RabbitMQ . :

• 每几百毫秒执行一次fsync
  
• 只有在检查每个节点的可用性的数据包的生存期（网络滴答）之后，才能检测到镜像。如果镜子减速或倒下，则会增加延迟。

Kafka依赖以下事实：如果消息存储在多个节点上，则可以在消息进入内存后立即对其进行确认。 因此，在同时发生故障的情况下，可能会丢失任何类型的消息（偶数acks = all ， min.insync.replies = 2 ）。

总体而言，Kafka表现出更好的性能，最初是为集群设计的。 如果需要，为了提高可靠性，可以将跟随者的数量增加到11个。 复制因子5和处于min.insync.replicas = 3的同步状态下的最小副本数将使消息丢失成为非常罕见的事件。 如果您的基础结构能够提供这样的复制速率和一定程度的冗余，则可以选择此选项。

RabbitMQ集群适用于小型队列。 但是即使是很小的队列也可以随着高流量而快速增长。 一旦队列变大，您将不得不在可用性和可靠性之间做出艰难的选择。 RabbitMQ集群最适合非典型情况，在这种情况下，RabbitMQ灵活性的优势胜过集群的任何劣势。

RabbitMQ的大队列漏洞的对策之一就是将它们分解为许多较小的对策。 如果您不需要整个队列的全部顺序，而只需要相关消息（例如，特定客户的消息），或者根本不需要所有消息，那么此选项是可以接受的：请查看我的Rebalanser项目以拆分队列（该项目仍处于早期阶段）。

最后，不要忘记RabbitMQ和Kafka的集群和复制机制中的许多错误。 随着时间的流逝，系统变得越来越成熟和稳定，但是没有一条消息能够100％受到保护，不会丢失！ 此外，数据中心还会发生大规模事故！

如果我错过任何事情，犯了一个错误或者您不同意任何观点，请随时发表评论或与我联系。

人们经常问我：“选择什么，Kafka还是RabbitMQ？”，“哪个平台更好？”。 事实是，这实际上取决于您的情况，当前的经验等。我不敢发表自己的意见，因为对于所有用例和可能的限制推荐任何一种平台都太简单了。 我写了这一系列文章，以便您形成自己的见解。

我想说两个系统都是该领域的领导者。 也许我有些偏颇，因为从我的项目经验中，我更倾向于欣赏诸如保证消息顺序和可靠性之类的东西。

我看到其他缺乏这种可靠性和无法保证订购的技术，然后看了RabbitMQ和Kafka-我了解这两种系统的不可思议的价值。