python-base--/tmp/kafka/KafkaPushTest11.py

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