具有动态数量的并行消费者的 Kafka 工作队列

Posted 2023-03-10

【中文标题】具有动态数量的并行消费者的 Kafka 工作队列

【英文标题】：Kafka work queue with a dynamic number of parallel consumers

【发布时间】：2018-12-24 14:44:39

【问题描述】：

我想用 Kafka 来“分工”。我想将工作实例发布到一个主题，并运行一组相同的消费者来处理它们。随着每个消费者完成其工作，它将从主题中提取下一个工作。每件作品只能由一个消费者处理一次。处理工作很昂贵，所以我需要很多消费者在很多机器上运行才能跟上。我希望消费者的数量根据需要增加和减少（我计划为此使用 Kubernetes）。

我发现了一种为每个消费者创建唯一分区的模式。这“分工”，但分区的数量是在创建主题时设置的。此外，主题必须在命令行上创建，例如

bin/kafka-topics.sh --zookeeper localhost:2181 --partitions 3 --topic divide-topic --create --replication-factor 1

...

for n in range(0,3):
    consumer = KafkaConsumer(
                     bootstrap_servers=['localhost:9092'])
    partition = TopicPartition('divide-topic',n)
    consumer.assign([partition])
    ...

我可以为每个消费者创建一个独特的主题，并编写自己的代码来为这些主题分配工作。这看起来很恶心，我仍然必须通过命令行创建主题。

具有动态数量的并行消费者的工作队列是一种常见的架构。我不能成为第一个需要这个的人。使用 Kafka 的正确方法是什么？

【参考方案1】：

您找到的模式是准确的。请注意，一旦创建了主题（有一些问题），也可以使用Kafka Admin API 和partitions can also be added 创建主题。

在 Kafka 中，划分工作并允许扩展的方式是使用partitions。这是因为在一个消费者组中，每个分区在任何时候都被一个消费者消费。

例如，您可以有一个具有 50 个分区的主题和一个订阅该主题的消费者组：

当吞吐量较低时，组中只能有几个消费者，他们应该能够处理流量。

当吞吐量增加时，您可以添加消费者，最多可达分区数（本例中为 50 个），以完成部分工作。

在这种情况下，50 个消费者是扩展的限制。消费者公开了许多指标（例如延迟），让您可以随时决定是否有足够的指标

【讨论】：

谢谢米凯尔。我不了解消费者组，以及消费者如何分配到分区。我会阅读这个。你能推荐一个链接/教程吗？

特别是：当我的partition很多，consumer很少的时候，消息是怎么分布的？在我的代码中，我手动将消费者分配给分区，因此他们只能在该分区中获取消息。但是，如果我不做分配，所有消费者都会收到所有消息。

虽然关于消费者群体的Kafka官方文档并不惊艳，但找到好的资源并不难。例如：blog.cloudera.com/blog/2018/05/…

关于分配，在一个组内，如果你的消费者少于分区，每个消费者将被分配几个分区。此外，如果您的消费者多于分区，则一些消费者将处于空闲状态并充当热备用，以防一些消费者死亡

【参考方案2】：

感谢 Mickael 为我指明了正确的方向。

https://www.safaribooksonline.com/library/view/kafka-the-definitive/9781491936153/ch04.html

Kafka consumers are typically part of a consumer group. When multiple
consumers are subscribed to a topic and belong to the same consumer group,
each consumer in the group will receive messages from a different subset of
the partitions in the topic.

https://dzone.com/articles/dont-use-apache-kafka-consumer-groups-the-wrong-wa,

Having consumers as part of the same consumer group means providing the
“competing consumers” pattern with whom the messages from topic partitions
are spread across the members of the group. Each consumer receives messages 
from one or more partitions (“automatically” assigned to it) and the same
messages won’t be received by the other consumers (assigned to different 
partitions). In this way, we can scale the number of the consumers up to the
number of the partitions (having one consumer reading only one partition); in
this case, a new consumer joining the group will be in an idle state without 
being assigned to any partition.

将工作分配给 3 个消费者的示例代码，最多 100 个：

bin/kafka-topics.sh --partitions 100 --topic divide-topic --create --replication-factor 1 --zookeeper localhost:2181

...

for n in range(0,3):
    consumer = KafkaConsumer(group_id='some-constant-group',
                     bootstrap_servers=['localhost:9092'])
    ...

【参考方案3】：

我认为，你在正确的道路上 -

这里涉及到一些步骤 -

创建 Kafka 主题并创建所需的分区。分区数是并行度的单位。换句话说，您运行这么多数量的消费者来处理工作。 如果扩展要求增加，您可以增加分区。但它带有重新分区等警告。请阅读有关新分区添加的 kafka 文档。 为消费者定义一个 Kafka 消费者组。 Kafka 会将分区分配给消费者组中可用的消费者并自动重新平衡。如果添加/删除了消费者，kafka 会自动进行重新平衡。 如果消费者被打包为 docker 容器，那么使用 kubernetes 有助于管理容器，尤其是对于多节点环境。其他工具包括 docker-swarm、openshift、Mesos 等。 Kafka 提供分区排序。 查看交付保证 - 根据您的使用案例，至少一次，准确一次。

或者，您可以使用 Kafka Streams APIS。 Kafka Streams 是一个客户端库，用于处理和分析存储在 Kafka 中的数据。它建立在重要的流处理概念之上，例如正确区分事件时间和处理时间、窗口支持以及简单而高效的管理和应用程序状态的实时查询。