很多事实，我们使用python 脚本，将需要处理的JSON 数据推送到 kafka 中间件中的案例 代码如下：

安装librdkafka

For RedHat and RPM-based distros, add this YUM repo and then do sudo yum install librdkafka-devel python-devel: http://docs.confluent.io/current/installation.html#rpm-packages-via-yum

from confluent_kafka import Consumer
from confluent_kafka import Producer
from confluent_kafka import KafkaError
#创建kafka 连接信息
producer_conf = {}
producer_conf['bootstrap.servers'] = 'localhost:9092'
producer = Producer(**producer_conf)

file_obj = open('1.json','r')
all_the_text = file_obj.readline()
file_obj.close()
#定义TOPIC信息
kafka_topic = "message_top"
try:
    producer.produce('topic',all_the_text,callback=self.del_callback)
    producer.poll(0)
            
except Exception as e:
    print('异常'+str(e))

producer.flush()

def del_callback(self,err,msg):
    if err:
        print(err)

producer.flush()  

这一句必须加 否则推送不过去