🥇了解消息代理。学习使用 ActiveMQ 和 Kafka 进行消息传递的机制。第 3 章卡夫卡

续小书的翻译：
了解消息代理
作者：Jakub Korab，出版商：O'Reilly Media, Inc.，出版日期：2017 年 9781492049296 月，ISBN：XNUMX。

上一个翻译部分：了解消息代理。学习使用 ActiveMQ 和 Kafka 进行消息传递的机制。第一章简介

第十九章

卡夫卡

Kafka 由 LinkedIn 开发，旨在解决传统消息代理的一些局限性，避免必须为不同的点对点交互设置多个消息代理，这在本书第 28 页的“向上和向外扩展”下进行了描述. 用例 LinkedIn 在很大程度上依赖于单向摄取大量数据，例如页面点击和访问日志，同时仍然允许多个系统使用这些数据，而不会影响生产者或其他消费者的生产力。事实上，Kafka 存在的原因是为了获得通用数据管道描述的那种消息传递架构。

鉴于此最终目标，自然会出现其他要求。卡夫卡应该：

速度极快
处理消息时提供更多带宽
支持发布者-订阅者和点对点模型
不要因为增加消费者而放慢速度。例如，随着目的地消费者数量的增加，ActiveMQ 中队列和主题的性能都会下降。
可横向扩展；如果一个持久化消息的代理只能以最大磁盘速度这样做，那么超越单个代理实例以提高性能是有意义的
限制对存储和重新检索消息的访问

为了实现这一切，Kafka 采用了一种重新定义客户端和消息代理的角色和职责的架构。 JMS 模型非常面向代理，代理负责分发消息，而客户端只需担心发送和接收消息。另一方面，Kafka 是以客户端为中心的，客户端具有传统代理的许多功能，例如向消费者公平分发相关消息，以换取极其快速和可扩展的代理。对于使用过传统消息系统的人来说，使用 Kafka 需要从根本上改变想法。
这一工程方向导致了消息传递基础设施的创建，与传统代理相比，该基础设施能够将吞吐量提高多个数量级。正如我们将看到的，这种方法需要权衡取舍，这意味着 Kafka 不适合某些类型的工作负载和已安装的软件。

统一目的地模型

为了满足上述要求，Kafka 在一种目的地下结合了发布-订阅和点对点消息传递 - 话题. 这让使用过消息系统的人感到困惑，其中“主题”一词指的是一种广播机制，从中（从主题）阅读是非持久的。正如本书简介中所定义的，Kafka 主题应该被视为一种混合目的地类型。

对于本章的其余部分，除非我们另有明确说明，否则术语“主题”将指代 Kafka 主题。

要充分了解主题的行为方式以及它们提供的保证，我们需要首先了解它们在 Kafka 中的实现方式。
Kafka 中的每个主题都有自己的日志。
向 Kafka 发送消息的生产者写入此日志，消费者使用不断向前移动的指针从日志中读取。 Kafka 会定期删除日志中最旧的部分，无论这些部分中的消息是否已被读取。 Kafka 设计的核心部分是代理不关心消息是否被读取——这是客户的责任。

术语“log”和“pointer”没有出现在卡夫卡文档. 此处使用这些众所周知的术语来帮助理解。

这种模型与ActiveMQ完全不同，ActiveMQ将来自所有队列的消息存储在同一个日志中，broker在读取消息后将消息标记为已删除。
现在让我们更深入地挖掘并更详细地查看主题日志。
Kafka 日志由几个分区组成（图3 1). Kafka 保证每个分区严格排序。这意味着以特定顺序写入分区的消息将以相同的顺序读取。每个分区都实现为一个滚动日志文件，其中包含 一个子集 其生产者发送到主题的所有消息的（子集）。默认情况下，创建的主题包含一个分区。分区的思想是Kafka进行水平扩展的中心思想。

图 3-1。卡夫卡分区

当生产者向 Kafka 主题发送消息时，它决定将消息发送到哪个分区。稍后我们将对此进行更详细的介绍。

阅读信息

想要读取消息的客户端管理一个名为 消费者群体，它指向抵消分区中的消息。偏移量是从分区开始处的 0 开始的增量位置。通过用户定义的 group_id 在 API 中引用的这个消费者组对应于 一个逻辑消费者或系统.

大多数消息传递系统使用多个实例和线程从目标读取数据以并行处理消息。因此，通常会有许多消费者实例共享同一个消费者组。

读取问题可以表示为：

主题有多个分区
多组消费者可以同时使用一个topic
一组消费者可以有多个独立的实例

这是一个不平凡的多对多问题。为了理解 Kafka 如何处理消费者组、消费者实例和分区之间的关系，让我们来看一系列逐渐复杂的读取场景。

消费者和消费者群体

让我们以一个具有一个分区的主题作为起点（图3 2).

图 3-2。消费者从分区读取

当消费者实例使用自己的 group_id 连接到该主题时，它会被分配一个读取分区和该分区中的偏移量。此偏移量的位置可在客户端中配置为指向最近位置（最新消息）或最早位置（最旧消息）的指针。消费者从主题中请求（轮询）消息，这导致它们从日志中顺序读取。
偏移位置定期提交回 Kafka 并作为消息存储在内部主题中 _consumer_offsets. 与常规代理不同，已读消息仍未删除，客户端可以倒回偏移量以重新处理已查看的消息。

当第二个逻辑使用者使用不同的 group_id 连接时，它管理独立于第一个指针的第二个指针（图3 3). 因此，Kafka 主题就像一个队列，其中有一个消费者，就像一个普通的发布-订阅 (pub-sub) 主题，多个消费者订阅，还有一个额外的好处，即所有消息都被存储并且可以被多次处理。

图 3-3。不同消费者组中的两个消费者从同一个分区读取

消费者组中的消费者

当一个消费者实例从分区读取数据时，它可以完全控制指针并处理消息，如上一节所述。
如果消费者的多个实例使用相同的 group_id 连接到具有一个分区的主题，那么最后连接的实例将获得指针的控制权，并且从那一刻起它将接收所有消息（图3 4).

图 3-4。同一个消费者组中的两个消费者从同一个分区读取

这种消费者实例数超过分区数的处理方式，可以认为是一种独占消费者。如果您需要消费者实例的“主动-被动”（或“热-暖”）集群，这可能很有用，尽管并行运行多个消费者（“主动-主动”或“热-热”）比消费者。待命。

与普通 JMS 队列的行为方式相比，上述消息分发行为可能令人惊讶。在这个模型中，发送到队列的消息将在两个消费者之间平均分配。

大多数情况下，当我们创建多个消费者实例时，我们这样做要么是为了并行处理消息，要么是为了提高读取速度，要么是为了增加读取过程的稳定性。由于一次只有一个消费者实例可以从一个分区中读取数据，那么这在 Kafka 中是如何实现的呢？

一种方法是使用单个消费者实例读取所有消息并将它们传递给线程池。虽然这种方法增加了处理吞吐量，但它增加了消费者逻辑的复杂性并且对增加读取系统的健壮性没有任何作用。如果消费者的一个副本由于电源故障或类似事件而宕机，则减法停止。

在 Kafka 中解决这个问题的规范方法是使用 bО更多的分区。

分区

分区是并行读取和扩展主题超出单个代理实例带宽的主要机制。为了更好地理解这一点，让我们考虑一种情况，其中有一个主题有两个分区并且一个消费者订阅了这个主题（图3 5).

图 3-5。一个消费者从多个分区读取

在这种情况下，消费者可以控制两个分区中与其 group_id 对应的指针，并开始从两个分区读取消息。
当将同一 group_id 的另一个消费者添加到该主题时，Kafka 将其中一个分区从第一个消费者重新分配给第二个消费者。之后，消费者的每个实例将从主题的一个分区中读取（图3 6).

要保证消息在 20 个线程中并行处理，至少需要 20 个分区。如果分区较少，您将只剩下无事可做的消费者，正如前面讨论独占消费者时所描述的那样。

图 3-6。同一个消费者组中的两个消费者从不同的分区读取

这种方案相比维护JMS队列所需的消息分发，大大降低了Kafka broker的复杂度。在这里您无需担心以下几点：

根据循环分配、预取缓冲区的当前容量或先前的消息（对于 JMS 消息组），哪个消费者应该接收下一条消息。
哪些消息被发送给哪些消费者，以及在失败的情况下是否应该重新传递它们。

Kafka 代理要做的就是在消费者请求时按顺序将消息传递给消费者。

然而，并行校对和重新发送失败消息的要求并没有消失——它们的责任只是从代理转移到客户端。这意味着您的代码中必须考虑到它们。

发送信息

该消息的生产者有责任决定将消息发送到哪个分区。要了解完成此操作的机制，我们首先需要考虑我们实际发送的到底是什么。

在 JMS 中，我们使用带有元数据（标头和属性）的消息结构和包含有效负载（有效负载）的主体，而在 Kafka 中，消息是 对“键值”. 消息负载作为值发送。另一方面，密钥主要用于分区并且必须包含 业务逻辑特定键将相关消息放在同一分区中。

在第 2 章中，我们讨论了需要由单个消费者按顺序处理相关事件的在线投注场景：

用户帐户已配置。
钱记入帐户。
下注是从账户中提取资金。

如果每个事件都是发布到主题的消息，那么自然键就是帐户 ID。
当使用 Kafka Producer API 发送消息时，它会传递给分区函数，根据消息和 Kafka 集群的当前状态，该函数返回消息应发送到的分区的 ID。此功能是通过 Partitioner 接口在 Java 中实现的。

这个界面看起来像这样：

interface Partitioner {
    int partition(String topic,
        Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);
}

Partitioner 实现对键使用默认的通用哈希算法来确定分区，如果没有指定键则使用循环法。在大多数情况下，此默认值效果很好。但是，将来您会想要自己编写。

编写自己的分区策略

让我们看一个示例，您希望在其中发送元数据以及消息有效负载。我们示例中的有效载荷是向游戏帐户存款的指令。指令是我们希望保证在传输时不被修改的东西，并且希望确保只有受信任的上游系统才能启动该指令。在这种情况下，发送和接收系统同意使用签名来验证消息。
在普通的 JMS 中，我们简单地定义一个“消息签名”属性并将其添加到消息中。但是Kafka并没有给我们提供传递元数据的机制，只有一个key和一个value。

由于该值是我们要保留其完整性的银行转账有效负载，因此我们别无选择，只能定义要在密钥中使用的数据结构。假设我们需要一个账户 ID 来进行分区，因为所有与账户相关的消息都必须按顺序处理，所以我们会得到以下 JSON 结构：

{
  "signature": "541661622185851c248b41bf0cea7ad0",
  "accountId": "10007865234"
}

因为签名的值会根据有效负载而变化，所以 Partitioner 接口的默认哈希策略不会可靠地对相关消息进行分组。因此，我们需要编写自己的策略来解析此键并对 accountId 值进行分区。

Kafka 包括校验和以检测存储中消息的损坏，并具有全套安全功能。即便如此，有时也会出现行业特定的要求，例如上述要求。

用户的分区策略必须确保所有相关消息最终都在同一个分区中。虽然这看起来很简单，但由于排序相关消息的重要性以及主题中分区数量的固定程度，要求可能会变得复杂。

一个主题中的分区数量会随着时间的推移而变化，因为如果流量超出最初的预期，它们可能会被添加。因此，消息键可以与它们最初发送到的分区相关联，这意味着在生产者实例之间共享一个状态。

另一个需要考虑的因素是跨分区的消息均匀分布。通常，密钥不会在消息中均匀分布，并且哈希函数不能保证一小部分密钥的消息公平分布。
请务必注意，无论您选择如何拆分消息，分隔符本身都可能需要重复使用。

考虑在不同地理位置的 Kafka 集群之间复制数据的需求。为此，Kafka 附带了一个名为 MirrorMaker 的命令行工具，用于从一个集群读取消息并将它们传输到另一个集群。

MirrorMaker 必须了解复制主题的键，以便在集群之间复制时保持消息之间的相对顺序，因为该主题的分区数在两个集群中可能不同。

自定义分区策略相对较少，因为默认哈希或循环法在大多数情况下效果很好。但是，如果您需要强顺序保证或需要从有效负载中提取元数据，那么您应该仔细研究分区。

Kafka 的可扩展性和性能优势来自于将传统代理的一些责任转移到客户端。在这种情况下，决定在并行工作的多个消费者之间分发可能相关的消息。

JMS 代理也需要处理这样的要求。有趣的是，通过 JMS 消息组（粘性负载平衡 (SLB) 策略的变体）实现的向同一消费者发送相关消息的机制也需要发送方将消息标记为相关。在 JMS 的情况下，代理负责将这组相关消息发送给众多消费者中的一个，并在消费者脱落时转移该组的所有权。

生产商协议

发送消息时，分区并不是唯一要考虑的事情。我们来看看Java API中Producer类的send()方法：

Future < RecordMetadata > send(ProducerRecord < K, V > record);
Future < RecordMetadata > send(ProducerRecord < K, V > record, Callback callback);

需要立即注意的是，这两个方法都返回了 Future，这表明发送操作不是立即执行的。结果是一条消息 (ProducerRecord) 被写入每个活动分区的发送缓冲区，并作为 Kafka 客户端库中的后台线程发送到代理。虽然这使事情变得非常快，但这意味着如果进程停止，没有经验的应用程序可能会丢失消息。

与往常一样，有一种方法可以以性能为代价使发送操作更可靠。这个缓冲区的大小可以设置为0，发送应用程序线程将被强制等待，直到消息传输到broker完成，如下：

RecordMetadata metadata = producer.send(record).get();

有关阅读消息的更多信息

阅读消息具有额外的复杂性，需要进行推测。与可以运行消息侦听器以响应消息的 JMS API 不同， 电子消费品 卡夫卡只轮询。让我们仔细看看这个方法 轮询（）用于此目的：

ConsumerRecords < K, V > poll(long timeout);

该方法的返回值是一个包含多个对象的容器结构 消费记录 来自可能的几个分区。 消费记录 本身是具有关联元数据的键值对的持有者对象，例如派生它的分区。

正如第 2 章所讨论的，我们必须牢记消息在成功或未成功处理后会发生什么，例如，如果客户端无法处理消息或中止消息。在 JMS 中，这是通过确认模式处理的。代理将删除成功处理的消息，或者重新传递原始或伪造的消息（假设使用了事务）。
Kafka 的工作方式非常不同。消息在校对后不会在代理中删除，失败时发生的情况是校对代码本身的责任。

正如我们所说，消费者组与日志中的偏移量相关联。与此偏移量关联的日志位置对应于要响应的下一条消息 轮询（）. 这个偏移量增加的时间点对于读取是决定性的。

回到前面讨论的阅读模型，消息处理包括三个阶段：

检索消息以供阅读。
处理消息。
确认消息。

Kafka 消费者自带一个配置选项 启用.auto.commit. 这是一个经常使用的默认设置，与包含单词“auto”的设置一样常见。

在 Kafka 0.10 之前，使用此选项的客户端将发送在下一次调用时读取的最后一条消息的偏移量 轮询（） 加工后。这意味着如果客户端已经处理过任何已经获取的消息但在调用之前意外销毁，则可以重新处理这些消息 轮询（）. 因为代理不会保留任何关于一条消息被阅读了多少次的状态，所以下一个检索该消息的消费者不会知道发生了什么不好的事情。这种行为是伪事务性的。仅当消息被成功处理时才提交偏移量，但如果客户端中止，代理将再次向另一个客户端发送相同的消息。此行为与消息传递保证一致”至少一次«。

在 Kafka 0.10 中，客户端代码已更改，以便根据配置由客户端库定期触发提交 自动提交间隔.ms. 此行为介于 JMS AUTO_ACKNOWLEDGE 和 DUPS_OK_ACKNOWLEDGE 模式之间。使用自动提交时，无论是否实际处理消息都可以提交消息——这可能发生在消费者速度较慢的情况下。如果消费者中止，消息将由下一个消费者获取，从提交的位置开始，这可能导致错过消息。在这种情况下，Kafka 并没有丢失消息，只是读取代码没有处理它们。

此模式与 0.9 版本具有相同的承诺：可以处理消息，但如果失败，可能不会提交偏移量，从而可能导致传递加倍。执行时获取的消息越多 轮询（），这个问题越多。

正如第 21 页的“从队列中读取消息”中所讨论的，在考虑故障模式的情况下，在消息传递系统中不存在一次性传递消息这样的事情。

在Kafka中，提交（commit）一个偏移量（offset）有两种方式：自动和手动。在这两种情况下，如果消息已处理但在提交之前失败，则可以多次处理消息。如果提交发生在后台并且您的代码在处理之前完成（可能在 Kafka 0.9 和更早版本中），您也可以选择根本不处理消息。

可以在Kafka consumer API中通过设置参数来控制手动offset commit过程 启用.auto.commit 为 false 并显式调用以下方法之一：

void commitSync();
void commitAsync();

如果您想“至少处理一次”消息，则必须手动提交偏移量 提交同步()通过在处理消息后立即执行此命令。

这些方法不允许消息在处理之前得到确认，但它们没有采取任何措施来消除潜在的处理延迟，同时提供事务性的外观。 Kafka 中没有事务。客户端无法执行以下操作：

自动回滚伪造的消息。消费者自己必须处理由有问题的有效负载和后端中断引起的异常，因为他们不能依赖代理重新传递消息。
在一个原子操作中将消息发送到多个主题。正如我们将很快看到的，对不同主题和分区的控制可以驻留在 Kafka 集群中的不同机器上，这些机器在发送时不协调事务。在撰写本文时，已经完成了一些工作，使 KIP-98 成为可能。
将阅读来自一个主题的一条消息与向另一个主题发送另一条消息相关联。同样，Kafka 的架构依赖于许多独立的机器作为一条总线运行，并且没有试图隐藏这一点。例如，没有允许您链接的 API 组件 消费者 и 制片人 在交易中。在 JMS 中，这是由对象提供的时间从中创建 消息生产者 и 消息消费者.

如果我们不能依赖事务，我们如何提供更接近传统消息系统提供的语义？

如果在消息被处理之前消费者的偏移量可能会增加，例如在消费者崩溃期间，那么消费者无法知道其消费者组在分配分区时是否错过了消息。因此，一种策略是将偏移量倒回到之前的位置。 Kafka 消费者 API 为此提供了以下方法：

void seek(TopicPartition partition, long offset);
void seekToBeginning(Collection < TopicPartition > partitions);

方法 寻找（） 可以与方法一起使用
offsetsForTimes（地图时间戳搜索） 回到过去某个特定时间点的状态。

隐含地，使用这种方法意味着很可能会再次读取和处理一些以前处理过的消息。为避免这种情况，我们可以使用第 4 章中描述的幂等读取来跟踪以前查看过的消息并消除重复项。

或者，您的消费者代码可以保持简单，只要消息丢失或重复是可以接受的。当我们考虑通常使用 Kafka 的用例（例如处理日志事件、指标、点击跟踪等）时，我们明白单个消息的丢失不太可能对周围的应用程序产生重大影响。在这种情况下，默认值是完全可以接受的。另一方面，如果您的应用程序需要发送付款，您必须仔细处理每条消息。这一切都取决于上下文。

个人观察表明，随着信息强度的增加，每条信息的价值都会降低。当以聚合形式查看时，大消息往往很有价值。

高可用性

Kafka 的高可用性方法与 ActiveMQ 的方法非常不同。 Kafka 是围绕横向扩展集群设计的，其中所有代理实例同时接收和分发消息。

Kafka 集群由运行在不同服务器上的多个代理实例组成。 Kafka 被设计为在普通的独立硬件上运行，其中每个节点都有自己的专用存储。不建议使用网络附加存储 (SAN)，因为多个计算节点会争用时间。Ыe 存储间隔并产生冲突。

卡夫卡是 永远在线 系统。许多大型 Kafka 用户从不关闭他们的集群，软件总是通过顺序重启进行更新。这是通过保证与以前版本的消息和代理之间交互的兼容性来实现的。

连接到服务器集群的代理动物园管理员，它充当配置数据注册表，用于协调每个代理的角色。 ZooKeeper 本身是一个分布式系统，通过建立信息复制来提供高可用性 法定人数.

在基本情况下，主题在 Kafka 集群中创建，具有以下属性：

分区数。如前所述，此处使用的确切值取决于所需的并行读取级别。
复制因子（factor）决定了集群中有多少代理实例应该包含这个分区的日志。

使用 ZooKeepers 进行协调，Kafka 尝试在集群中的代理之间公平地分配新分区。这是由充当控制器的单个实例完成的。

在运行时 对于每个主题分区 控制者 为经纪人分配角色 领导人 （领导者，大师，主持人）和 追随者 （追随者、奴隶、下属）。代理作为该分区的领导者，负责接收生产者发送给它的所有消息并将消息分发给消费者。当消息发送到主题分区时，它们将被复制到充当该分区的跟随者的所有代理节点。包含分区日志的每个节点称为 复制品. 经纪人可以充当某些分区的领导者和其他分区的追随者。

包含领导者持有的所有消息的追随者称为 同步副本 （处于同步状态的副本，同步副本）。如果充当分区领导者的代理发生故障，则该分区的任何最新或同步的代理都可以接管领导角色。这是一个令人难以置信的可持续设计。

生产者配置的一部分是参数确认，它决定了在应用程序线程继续发送之前有多少副本必须确认（acknowledge）收到一条消息：0、1 或全部。如果设置为所有，那么当接收到一条消息时，领导者将在收到来自主题设置定义的多个线索（包括它自己）的记录的确认（确认）后立即向生产者发送确认 最小同步副本数 （默认 1）。如果无法成功复制消息，则生产者将抛出应用程序异常（没有足够的副本 или 追加后副本不足).

一个典型的配置创建一个主题，其复制因子为 3（每个分区 1 个领导者，2 个追随者）和参数 最小同步副本数 设置为 2。在这种情况下，集群将允许管理主题分区的代理之一关闭而不影响客户端应用程序。

这让我们回到了熟悉的性能和可靠性之间的权衡。复制的发生是以等待来自追随者的确认（确认）的额外等待时间为代价的。虽然，因为它是并行运行的，所以复制到至少三个节点与两个节点具有相同的性能（忽略网络带宽使用的增加）。

通过使用这种复制方案，Kafka 巧妙地避免了将每条消息物理写入磁盘的需要 同步（）. 生产者发送的每条消息都将写入分区日志，但如第 2 章所述，写入文件最初是在操作系统的缓冲区中完成的。如果这条消息被复制到另一个 Kafka 实例并在它的内存中，领导者的丢失并不意味着消息本身丢失了——它可以被一个同步的副本接管。
拒绝执行操作 同步（） 意味着 Kafka 接收消息的速度与将消息写入内存的速度一样快。相反，避免将内存刷新到磁盘的时间越长越好。因此，为 Kafka 代理分配 64 GB 或更多内存的情况并不少见。这种内存使用意味着单个 Kafka 实例可以轻松地以比传统消息代理快数千倍的速度运行。

Kafka 也可以配置为应用操作 同步（） 消息包。由于 Kafka 中的所有内容都是面向包的，因此它实际上适用于许多用例，并且对于需要非常强大的保证的用户来说是一个有用的工具。 Kafka 的许多纯粹性能来自于作为数据包发送到代理的消息，并且这些消息是使用顺序块从代理中读取的零拷贝操作（不执行将数据从一个内存区域复制到另一个内存区域的操作）。后者是一个很大的性能和资源增益，只有通过使用定义分区方案的底层日志数据结构才有可能。

与单个 Kafka 代理相比，Kafka 集群中的性能可能要好得多，因为主题分区可以横向扩展到许多单独的机器上。

结果

在本章中，我们了解了 Kafka 架构如何重新构想客户端和代理之间的关系，以提供极其强大的消息传递管道，其吞吐量比传统消息代理高出许多倍。我们已经讨论了它用于实现此目的的功能，并简要介绍了提供此功能的应用程序的体系结构。在下一章中，我们将研究基于消息传递的应用程序需要解决的常见问题，并讨论处理这些问题的策略。我们将在本章结尾概述如何一般地讨论消息传递技术，以便您可以评估它们对您的用例的适用性。

上一个翻译部分：了解消息代理。学习使用 ActiveMQ 和 Kafka 进行消息传递的机制。第1章

翻译完成： tele.gg/middle_java

待续...

只有注册用户才能参与调查。登录拜托

您的组织使用 Kafka 吗？

是
没有
以前用过，现在不用
我们打算用

38 位用户投票。 8 名用户弃权。

来源： habr.com

了解消息代理。 学习使用 ActiveMQ 和 Kafka 进行消息传递的机制。 第 3 章卡夫卡