python3 操作kafka 創建生產者、消費者示例

1、kafka  生產者端的程序如下（kafka 版本: kafka 1.3.5）：

查看服務端的 kafka topics指令： kafka-topics.sh --zookeeper  localhost:2181 --list

from kafka import KafkaProducer

kafkaClient=KafkaProducer(bootstrap_servers='xx.xx.xx.203:9092')

def start_producer(msg):

    kafkaClient.send('t_jg_lt_producer',msg.encode())
    kafkaClient.flush()
    print("客戶端發送完成：%s"%msg)


if __name__=='__main__':
    msg="你好嗎，我的小伙伴，miomiao"
    start_producer(msg)

2、kafka 消費者端：

from kafka import KafkaConsumer

def start_consumer():
    kafkaConsumer=KafkaConsumer('t_jg_lt_producer',bootstrap_servers='xx.xx.xx.203:9092')
    for msg in kafkaConsumer:
        #print('接收到的信息為：',msg)
        print("轉換后的value：",msg.value.decode())


if __name__=='__main__':
    start_consumer()