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

Question

我希望使用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应该如何正确实现？

- opus111

Answer 1

你找到的模式是准确的。请注意，主题也可以使用Kafka管理API创建，并且一旦创建了主题（有一些陷阱），还可以添加分区。

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

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

- 当吞吐量较低时，你可以只有少数消费者在组内，并且它们应该能够处理流量。 - 当吞吐量增加时，你可以添加消费者，最多达到分区数（在此示例中为50），以处理部分工作。

在这种情况下，消费者数量的限制是50个。消费者会暴露一些指标（例如滞后），让您可以随时决定是否有足够的消费者。

- Mickael Maison

---

谢谢Mickael。我不理解消费者组以及消费者如何分配到分区。我会自己去了解。你能否建议一个链接或教程？ - opus111

特别是当我有许多分区和少量消费者时，消息如何分配？在我的代码中，我手动将消费者分配到分区，因此他们只会收到该分区的消息。但是，如果我不进行分配，则所有消费者都会收到所有消息。 - opus111

尽管官方的Kafka文档关于消费者组并不是很好，但是很容易找到好的资源。例如：https://blog.cloudera.com/blog/2018/05/scalability-of-kafka-messaging-using-consumer-groups/ - Mickael Maison

关于分配，如果在一个组内你的消费者少于分区数，每个消费者将被分配多个分区。同样，如果你的消费者多于分区数，一些消费者将处于空闲状态，并作为热备份以防一些消费者死亡。 - Mickael Maison

Answer 2

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

以下是一些步骤 -

1. 创建Kafka主题并创建所需的分区。分区数是并行性的单位。换句话说，您可以运行这么多个消费者来处理工作。
2. 如果扩展要求增加，则可以增加分区。但是，它会带来一些注意事项，例如重新分区。请阅读有关新分区添加的kafka文档。
3. 为消费者定义Kafka Consumer组。 Kafka将向可用的消费者组分配分区并自动重新平衡。如果添加/删除了消费者，则kafka会自动进行重新平衡。
4. 如果将消费者打包为docker容器，则使用kubernetes有助于管理容器，特别是对于多节点环境。其他工具包括docker-swarm，openshift，Mesos等。
5. Kafka为分区提供排序。
6. 检查交付保证 - 根据您的用例，至少一次，仅一次。

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

- Mac

Answer 3

由于您有一个慢消费者用例，因此 Confluent 的并行消费者 (PC) 是非常适合的。PC 直接解决了这个问题，通过按键对输入分区进行子分区，并并行处理每个键。因此，处理可以花费任意长的时间。它还跟踪每个记录的确认。请查看 Parallel Consumer GitHub（顺便说一下，这是开源的，我是作者）。

- Antony Stubbs

Answer 4

感谢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'])
    ...

- opus111