沃尔玛如何利用Kafka 处理每日数万亿条消息

沃尔玛在私有云和公共云环境中拥有大规模的 Apache Kafka 部署，其消费应用数量超过 25,000。

该部署每天处理数万亿条 Kafka 消息，可用性高达 99.99%。它支持以下关键用例：

• 数据移动

• 事件驱动的微服务

• 流分析

在沃尔玛这样的规模下，Kafka 设置必须能够应对突发的流量高峰。此外，由于消费应用程序应用程序使用多种编程语言开发，所有这些应用程序都必须遵循一套最佳实践，以保持一致的可靠性和质量标准。

沃尔玛规模化使用 Kafka 面临的挑战

让我们首先了解沃尔玛面临的主要挑战。

1 – 消费应用程序重新平衡

最常见的问题之一与消费者应用程序的重新平衡有关。

那么，什么会触发 Kafka 中的消费者应用程序重新平衡呢？

这通常是由消费者组内的消费者应用程序实例数量发生变化而引起的。

可能存在多种情况，例如：

• 消费应用程序进入或离开消费组。这可能由 Kubernetes 部署、滚动重启或自动缩扩容引起。每当发生这种情况，Kafka 都需要在消费应用程序之间重新分配分区。

• Kafka 代理可能认定某个消费应用程序已失效。如果代理在设定的会话超时时间内未收到消费应用程序的心跳信号，就会做出这种判断。消费应用程序的 JVM 退出或经历长时间的垃圾回收暂停都可能导致这种情况。

• Kafka 代理可能判定某个消费应用程序已卡住，从而触发重新平衡。如果消费应用程序轮询下一批记录的时间超过阈值，代理会将其标记为卡住。这通常发生在处理上一批记录耗时过长的情况下。

重新平衡消费应用程序是为了确保分区均匀分布。然而，在电子商务这样需要近实时处理的环境中，重新平衡可能会导致服务中断和延迟增加。

2-毒丸信息

Kafka 中的"毒丸"消息是指消费应用程序无法成功处理的消息。这种情况可能由多种原因引起，例如：

• **数据格式错误：**消息负载可能不符合预期格式。比如，无效的 JSON 或缺少必填字段。这可能导致消费应用程序在处理时抛出异常。

• **意外数据：**消息内容在语法上可能正确，但在语义上不合规。换言之，它可能违反了某些业务规则。

• **消费应用程序代码中的错误：**如果处理消息的代码存在缺陷（如空指针异常），处理将会失败。

当消费应用程序遇到这类消息时，它将无法处理并抛出异常。默认情况下，消费应用程序会重新向代理请求同一批消息。由于毒丸消息仍在该批次中，消费应用程序将再次无法处理它，这个循环会无限持续。

结果，消费应用程序被卡在这条问题消息上，无法继续处理分区中的其他消息。这种情况类似于网络中的"队头阻塞"问题。

3 - 成本问题

Kafka 主题中的分区数与可并行读取该主题的最大消费应用程序数量之间存在强耦合。这种耦合在扩展消费应用程序以处理更高吞吐量时会导致成本增加。

举例来说，假设您有一个包含 10 个分区的 Kafka 主题（Topic），以及 10 个读取该主题数据的消费应用程序实例。如果传入消息速率增加，而消费应用程序跟不上（即消费应用程序延迟开始增加），可能需要通过添加更多实例来扩展消费应用程序。

然而，一旦单个组中有 10 个消费应用程序（每个分区一个），再添加消费应用程序就无济于事了。这是因为 Kafka 不会将同一组中的多个消费应用程序分配给一个分区。要允许组中有更多消费应用程序，唯一的方法是增加主题中的分区数量。

不过，增加分区数量也会带来挑战和成本：

• Kafka 对每个代理的分区数量有建议限制（例如，每个代理 4000 个分区）。如果继续增加分区，可能会达到此限制，从而需要将 Kafka 代理扩展到更大的实例——即使代理有足够的资源来处理当前负载。扩展到更大的代理实例成本高昂。

• 增加分区需要 Kafka 团队、生产者和消费应用程序团队之间的协调。在拥有数千条 Kafka 管道的大型组织中，这种协调开销巨大。

• 分区数量越多，意味着打开的文件句柄越多、内存使用量越大，以及 Kafka 代理上的线程越多。这会导致更高的资源利用率和成本。

  
  

  
  

沃尔玛的消息代理服务（ Messaging Proxy Service—MPS）

为了克服以上提到的挑战，沃尔玛工程团队设计了消息代理服务（MPS）。

MPS 旨在解除 Kafka 消息消费与 Kafka 基于分区模型的限制。它的工作原理如下：

• MPS 作为 Kafka 代理与实际消息消费应用程序之间的中介。它从 Kafka 分区读取消息，并将其存入独立的内存队列。

• 消费应用程序不直接从 Kafka 读取数据，而是通过 HTTP/REST 从 MPS 接收消息。这使得消费应用程序可以独立于 Kafka 分区数量进行扩展。

• MPS 确保按键顺序处理消息，处理消费应用程序故障，并管理向 Kafka 的偏移提交。

下图展示了 MPS 及其所有组件的详细设计：

现在让我们更详细地了解 MPS 的各个组件。

MPS（消息代理服务）的组件介绍。这是对MPS系统各个部分的详细解释，包括其功能和工作原理。这些组件共同构成了MPS的整体架构，用于高效处理和分发Kafka消息。

Reader Thread

这是一个从 Kafka 读取消息的单线程。

它将从 Kafka 代理接收的消息写入一个名为"PendingQueue"的有界队列。当 PendingQueue 达到其最大容量时，读取器线程会暂停从 Kafka 读取。

这种机制形成了一种背压，防止队列在写入线程处理不及时时无限增长。

有界缓冲队列 (Pending Queue)

这是位于读取线程和写入线程之间的队列，设有最大容量限制以防止过度消耗内存。

PendingQueue 使读取线程和写入线程能够以不同速度运作。读取线程可按 Kafka 提供的速率读取消息，而写入线程则可按自身节奏处理消息。这种设计实现了两个线程之间的解耦，提高了系统的灵活性和效率。

顺序迭代器（Order Iterator）

该组件确保具有相同密钥的消息按照从 Kafka 接收的顺序进行处理。它会检查 PendingQueue 中的消息，如果发现之前已有一条相同密钥的消息正在处理，则会跳过当前消息。这样，在任何给定时刻，每个密钥最多只有一条消息由写入线程处理。

Writer Threads

这些是一组线程池，负责从PendingQueue中获取消息，并通过HTTP POST请求将其发送给消费者应用程序。

若POST请求失败，写入线程会进行多次重试。如果重试次数耗尽，或消费者应用程序返回特定的HTTP代码，写入线程会将该消息转移至死信队列（DLQ）。

此外，写入线程还参与偏移量管理。它们会更新一个共享数据结构，用于追踪已处理消息的偏移量。

偏移提交线程（Offset Commit Thread）

这是一个独立的线程，它会定期（如每分钟）唤醒，并使用 Kafka 消费者 API 提交已处理消息的 Kafka 偏移量。

该线程检查写入线程更新的共享数据结构，然后为每个分区提交最新的连续偏移量。例如，如果某个分区已处理的消息偏移量为 1、2、3 和 5，它将提交偏移量 3（因为 4 缺失）。

通过定期提交偏移量，MPS 向 Kafka 报告已成功处理的消息。如果 MPS 发生崩溃或重启，它将从最后提交的偏移量开始消费消息，从而避免重复处理。

消费应用程序服务的REST API

这是实际消息消费应用程序从 MPS 接收消息所需实现的规范。

它定义了 MPS 写入器线程将发送的 HTTP POST 请求格式（包括标头和正文等）。同时，它还指定了消费应用程序可返回的不同 HTTP 响应代码及其含义。

请参阅下表，其中显示了 API 规范：

来源：沃尔玛科技博客

MPS 的应用

MPS 被实现为 Kafka Connect 接收器连接器。

作为参考，Kafka Connect 是一个用于将 Kafka 与外部系统（如数据库、键值存储、搜索索引和文件系统）连接的框架。它提供了一种标准方法来定义将数据移入和移出 Kafka 的连接器。

下图展示了 Kafka Connect 的流转视图

通过将 MPS 实现为 Kafka Connect 接收器连接器，开发人员能够利用 Kafka Connect 框架提供的多项功能，包括：

• 多租户：Kafka Connect 支持在单个集群上运行多个连接器，使单个 MPS 部署能够服务多个消费应用程序（租户）。

• 死信队列 (DLQ) 处理：Kafka Connect 内置了对无法处理消息的支持。

• 偏移提交：Kafka Connect 提供了用于提交偏移的 API，MPS 在其偏移提交线程中使用这些 API。

• 可扩展性：Kafka Connect 的设计具有可扩展性和容错性，MPS 通过构建在其之上继承了这些特性。

此外，消费应用程序服务（处理消息的应用程序）被设计为无状态的，意味着它们不在本地维护任何持久状态。所需的任何状态要么随消息传递，要么存储在外部数据库中。

这种无状态设计使得这些服务能够根据消息量的变化在 Kubernetes 中灵活地扩展或缩减。需要处理更多消息时，Kubernetes 只需启动更多消费应用程序服务实例；当传入消息减少时，Kubernetes 可以终止一些实例以节省资源。

值得注意的是，消费应用程序服务的这种扩展与 MPS 和 Kafka 无关。无论运行多少消费应用程序服务实例，MPS 都会持续从 Kafka 读取数据并将消息传递给消费应用程序服务。

需要考虑的事项

以下是根据沃尔玛实施的 MPS 解决方案值得考虑的其他几点。

1 - MPS 的重新平衡

MPS 本质上也是一个 Kafka 消费应用程序。它从 Kafka 主题读取消息，并通过 REST 端点将其提供给消费应用程序。

与其他 Kafka 消费应用程序一样，MPS 实例数量变化时也会触发重新平衡。然而，MPS 的设计似乎能够优雅地处理这一过程。其中的关键在于读取器线程（轮询 Kafka）和写入器线程（向 REST 消费应用程序发送消息）的分离。

只要 MPS 在重新平衡后能迅速恢复，REST 消费应用程序就应该能够持续处理消息，不会出现明显延迟。MPS 的设计还包括读取线程和写入线程之间的有界缓冲区（PendingQueue），这有助于平滑 MPS 从 Kafka 读取速率的任何短暂波动。

2 - REST 的选择

MPS 调用消费应用程序实例所公开的 REST API。值得注意的是，它选择了 REST 而非 gRPC 等其他协议。

这一选择可能源于 REST 的简单性，以及几乎所有编程语言和框架对 REST 的广泛支持。

3 - 潜在的复杂性增加

尽管 MPS 解决了一些问题，但它也为系统引入了一个额外的层。

现在，Kafka 和消费应用程序之间多了一个代理服务。这意味着需要额外开发、部署、监控和维护更多的组件。

结论

MPS 的实施帮助沃尔玛实现了一些关键的改进：

• 大多数重新平衡问题已得到解决，仅在极少数重启或网络问题时才会发生。MPS 的读取器线程在规定时间内将所有轮询的消息放入 PendingQueue，有效防止了 Kafka 代理误判消费应用程序卡住而触发重新平衡。

• 毒丸消息的处理得到显著改善。通过 MPS，消费应用程序服务能够识别毒丸消息，并使用特定的 HTTP 返回代码（600 和 700）通知 MPS。

• MPS 在成本节省方面表现出色。首先，消费应用程序服务现在采用无状态设计，可以根据需求在 Kubernetes 中灵活扩展，无需提前扩容。其次，Kafka 集群的扩展现在基于吞吐量而非分区数量，进一步优化了资源利用。