Kafka-生产者/消费者

2022-04-27 20:01:37 阅读：147 来源： 互联网

标签：消费者 producer 生产者 value kafka send msg message Kafka

1 安装Kafka库

# 推荐安装
pip install kafka-python

# 不兼容python3.8
pip install kafka

2 生产者

import json
from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers='192.168.2.230:9092')

msg_dict = {
    "operatorId":"test",
    "terminalId":"123",
    "terminalCode":"123",
    "terminalNo":"1"
}
msg = json.dumps(msg_dict).encode() #这里加了encode 进行了编码，不然会报错：使用KafkaProducer的send函数时，会报assert type(value_bytes) in (bytes, bytearray, memoryview, type(None))错误原因：send函数的value_bytes是str类型
producer.send('tqs-admin-event-1', msg)
producer.close()
print("结束")

2 消费者

from kafka import KafkaConsumer

consumer = KafkaConsumer('test',
                         bootstrap_servers=['172.21.10.136:9092'])
                         
for message in consumer:
    print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition,
                                          message.offset, message.key,
                                          message.value))

标签：消费者,producer,生产者,value,kafka,send,msg,message,Kafka
来源： https://www.cnblogs.com/shukeshu/p/16200322.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

Kafka-生产者/消费者