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()