python连接kafka生产者发送消息
通过pip install kafka-python安装第三方工具
再导入相应的方法就可以连接kafka进行消息发送了。
from kafka import KafkaProducer, KafkaConsumer
import jsonproducer = 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 jsonproducer = 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之后，就发送成功了。