更详细的可以参考：https://zhuanlan.zhihu.com/p/279784873
https://blog.csdn.net/weixin_35688430/article/details/111292744
生产者：producer
生产者kafka有对应的三方库可以支撑去进行信息发送。
from kafka import KafkaProducer
import json
def kafka_sendMsg(topic,bootstrap_servers,key,msg,header) :
producer = KafkaProducer(value_serializer= lambda v :json.dumps(v).encode('utf-8'),bootstrap_servers=['127.0.0.1:9092'])
消费者：consumer
消费者，同样可以通过topic和服务器名称，去获取对应的数据。
from kafka import KafkaconsumerMsg(topic,bootstrap_servers)
import json
def kafka_sendMsg(topic,bootstrap_servers,key,msg,header) :
consumer= KafkaConsumer(topic,bootstrap_servers=bootstrap_servers)
for msg in consumer:
recv ="%s:%d%d: key=%s value=%s" %(msg.topic,msg.partition,msg.offset,msg.key, msg.value)
使用kafka
def kafka():
topic = "AA1.reaply.info" #kafka的主题名称
bootstrap_servers = "127.0.0.1" #kafka连接
kafkaMsg = kafka_msg_mode.getModel() #字典，获取对应的取值
key = bytes(kafkaMsg["msgkey"]+self.feature.loginUserNo,encoding='utf-8') #kakfa的key
msg = json.loads(topic_str_info) #kafka的value
header =["Headers"]
kafka_sendMsg(topic,bootstrap_servers,key,msg,header) #向kafka中间件发送消息
