Kafka Consumer(Python threading)

Posted 2020-09-30

import threading
from kafka import KafkaConsumer

threads = []

class MyThread(threading.Thread):
    def __init__(self, threadName, keyName):
        threading.Thread.__init__(self)
        self.threadName=threadName
        self.keyName=keyName

    def run(self):
        receiveinfo(self.threadName, self.keyName)

def receiveinfo(threadName, keyName):
    consumer = KafkaConsumer(‘test‘, bootstrap_servers=‘192.168.1.10:9092‘)
    for msg in consumer:
        if msg.key==keyName:
            if msg.value=="exit": break
            print("("+threadName+")"+ " " + msg.value)
try:
    t1=MyThread("T1","Thread-1")
    threads.append(t1)
    t2=MyThread("T2","Thread-2")
    threads.append(t2)
    t3=MyThread("T3","Thread-3")
    threads.append(t3)

    for t in threads:
        t.start()

    for t in threads:
        t.join()

    print ("exit program with 0")
except:
   print ("Error: failed to run consumer program")