python - 如何在 python 中扩展 Kafka Consumers？

Question

这可能有多个问题，所以请耐心等待。我仍在寻找使用 Kafka 架构的正确方法。我知道一个主题的分区是由消费者划分的。

消费者究竟是什么？现在，我正在考虑编写一个充当消费者的守护进程 python 进程。当消费者消费来自 Kafka 的消息时，我必须完成一项任务。这是一项艰巨的任务，所以我正在创建并发运行的子任务。我可以在同一台机器上有多个消费者(python 脚本)吗？

我有多个正在处理的微服务，所以每个微服务都有自己的使用者？

当负载增加时，我必须扩展消费者。我想产生一台充当另一个消费者的新机器。但我只是觉得我在这里做错了，觉得必须有更好的方法。

你能告诉我你是如何根据负载扩展你的消费者的吗？如果我需要增加我的消费者，我是否必须增加我的主题分区？我如何动态地做到这一点？当产生的消息较少时，我可以减少分区吗？最初有多少个分区是理想的？

并请提出一些可以遵循的良好做法。

这是我正在使用的消费者脚本

while True:
    message = client.poll(timeout=10)#client is the KafkaConsumer object
    if message is not None:
        if message.error():
            raise KafkaException(message.error())
        else:
            logger.info('recieved topic {topic} partition {partition} offset {offset} key {key} - {value}'.format(
                topic=message.topic(),
                partition=message.partition(),
                offset=message.offset(),
                key=message.key(),
                value=message.value()
            ))
            #run task

Answer 1

Can I have multiple consumers(python scripts) on the same machine?

是的。不过，您也可以拥有 Python 线程。

如果您不消费多个主题，则不需要多个消费者。

What exactly are consumers?

请随意阅读 Apache Kafka 站点...

each microservice has its own consumer?

每个服务是否运行相似的代码？好的。

I thought of spawning a new machine

在一台机器上生成应用程序的新实例。监控 CPU 和内存以及网络负载。在正常处理下至少其中一台机器超过 70% 之前，不要购买新机器。

Do I have to increase my partitions in topics if I need to increase my consumers?

一般来说，是的。一个消费者组中的消费者数量受订阅主题中分区数量的限制。

Can I decrease the partitions when there are fewer messages produced?

不能。分区不能减少

When the load increases I have to scale the consumers

不必要。增加的负载是不断上升还是有波动？如果可变，那么您可以让 Kafka 缓冲消息。消费者将尽可能快地进行轮询和处理。

您需要定义消息从生产者到达主题后需要多长时间处理的 SLA。

How many partitions are ideal initially?

这方面有很多文章，具体取决于您自己的硬件和应用程序要求。只需记录每条消息，您就可以拥有数千个分区......

When the consumer consumes a message from Kafka, there is a task that I have to complete

听起来你可能想看看 Celery，不一定只是 Kafka。您也可以 look at Faust用于 Kafka 处理