python连接kafka生产者发送消息

通过pip install kafka-python安装第三方工具
再导入相应的方法就可以连接kafka进行消息发送了。

from kafka import KafkaProducer, KafkaConsumer import json  producer = KafkaProducer(bootstrap_servers=['xxx.xxx.xxx.xxx:9092','xxx.xxx.xxx.xxx:9092'],                          security_protocol='SASL_PLAINTEXT',                          sasl_mechanism='SCRAM-SHA-512',                          sasl_plain_username = '鉴权账户',                          sasl_plain_password = '鉴权password',                          value_serializer = lambda m: json.dumps(m).encode('ascii')) message = {"test":"test"} producer.send('topicname', value = message) producer.flush() 

同时发送msg和key时，有时会报错key_bytes不属于字节类型，于是就手动把json内容改成字节型，再发送

from kafka import KafkaProducer, KafkaConsumer import json  producer = KafkaProducer(bootstrap_servers=['xxx.xxx.xxx.xxx:9092','xxx.xxx.xxx.xxx:9092'],                          security_protocol='SASL_PLAINTEXT',                          sasl_mechanism='SCRAM-SHA-512',                          sasl_plain_username = '鉴权账户',                          sasl_plain_password = '鉴权password') message = {"test":"test"} k = {"test":"test"}  msg_bytes =  json.dumps(message).encode('ascii') k_bytes = json.dumps(k).encode('ascii')  producer.send('topicname', key = k_bytes, value = msg_bytes) producer.flush() 

遇到过send发送消息超过60000ms的错误，这个问题需要加上鉴权，并且确认topic是正确的。我是加了鉴权，即用账户和密码，并且改正了topic之后，就发送成功了。