有没有一种方法可以在 python 中使用多处理从 kafka 主题中读取大量消息？答案

【问题标题】：Is there a way I can read large number of messages from a kafka topic using multiprocessing in python?有没有一种方法可以在 python 中使用多处理从 kafka 主题中读取大量消息？
【发布时间】：2021-02-18 16:26:19
【问题描述】：

我正在尝试从最早的偏移量读取 kafka 主题，然后通过 python 脚本墓碑某些记录。由于消息数量巨大（百万+），我想利用多处理来使脚本在使用消息时更快。这是脚本中的一个 sn-p：

    from kafka import KafkaConsumer

    def cleanup_kafka_topic(self, env):
    # Declarations
    consumer = KafkaConsumer(<topic_name>, group_id=<some_group>),
                             bootstrap_servers=[<kafka_host:kafka_port>],
                             auto_offset_reset='earliest', enable_auto_commit=True)
    # Clean-up logic
    for msg in consumer:
        # Do something with the msg

我正在使用 Kafka-python。

【问题讨论】：

因为这是一种I/O操作，我认为多线程比多处理更适合这种情况，你可以使用concurrent.futures模块进行测试
如果你想要多个异步消费者，你可以使用 aiokafka

标签： python multithreading apache-kafka multiprocess kafka-python

【解决方案1】：

Kafka 消费者不是线程安全的（请参阅此处的线程安全部分：https://pypi.org/project/kafka-python/）。加快速度的方法是在您的主题上拥有多个分区并增加消费者的数量（都具有 sae 消费者组标识符）。如果您有 N 个分区，则最多可以有 N 个消费者（每个分区最多可以有 1 个消费者）。 Kafka 将在您的消费者上升或下降时负责分配和重新分配分区，以便您可以按需扩展（例如，通过观察分区上的滞后）。请注意，根据文档，这需要使用更新的（0.9+）kafka 代理。

【讨论】：