python kafka多线程消费数据

时间:2021-08-11
本文章向大家介绍python kafka多线程消费数据,主要包括python kafka多线程消费数据使用实例、应用技巧、基本知识点总结和需要注意事项,具有一定的参考价值,需要的朋友可以参考一下。

1、打印每个线程id,满足预期,开启了8个线程,每个线程号都不一样;

2、查看kafka状态,也能满足预期,每个分区的消费者id都是不一样的,下面第二个图是开启一个消费者时的状态,每个分区的消费者id都是相同的;对比之下能满足需求;

3、相关代码如下:

from kafka import KafkaConsumer
import time, threading
from concurrent.futures import ThreadPoolExecutor


class ThreadKafka(object):

    def __init__(self):
        self.threadPool = ThreadPoolExecutor(max_workers=8, thread_name_prefix="threading_")
        self.hosts = ["ip:port", "ip:port", "ip:port"]

    def operate(self):
        consumer = KafkaConsumer("topic_name", bootstrap_servers=self.hosts, group_id="group_id_name")
        print(threading.current_thread().name)
        for i in consumer:
            print(i)
            time.sleep(1)

    def main(self):
        for i in range(8):
            self.threadPool.submit(self.operate, )


if __name__ == '__main__':
    cla = ThreadKafka()
    cla.main()

  

原文地址:https://www.cnblogs.com/qiaoer1993/p/15128595.html