1. 程式人生 > 其它 >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()