如何让kafka-python消费者并行

Question

我有一组kafka-python消费者，他们持续并行地使用不同的kafka主题。

我的问题是，如何使用单个python脚本并行启动消费者？管理(启动/停止/监视)这些消费者的最佳方式是什么？

如果我写下ex: run.py

import consumer1, consumer2, consumer3

consumer1.start()
consumer2.start()
consumer3.start()

它只是挂起在consumer1.start()上，因为脚本不返回任何值并保持运行。

Answer 1

您可以为每个使用者提供不同的线程，以便并行使用消息。例如，您可以拥有：

consumer1_thread = threading.Thread(target=consumer1.start, args=())
consumer2_thread = threading.Thread(target=consumer2.start, args=())
consumer3_thread = threading.Thread(target=consumer3.start, args=())
consumer1_thread.start()
consumer2_thread.start()
consumer3_thread.start()

您可以并行查看每个线程的日志，如果需要，还可以编写一些逻辑来停止单个线程。