import json import time from kafka import KafkaProducer producer = KafkaProducer(value_serializer=lambda v: json.dumps(v).encode('utf-8'), bootstrap_servers=['192.168.36.199:9092']) index = 0 while True: time.sleep(2) # 每个5秒推送一条数据 producer.send('tag_atom_topic', json.dumps({ "msg": "nihao" }), partition=0) # 发送的topic为test print("===")