python kafka多執行緒消費資料
阿新 • • 發佈:2021-08-11
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()