使用python发送和接口kafka

胡八万

发布于 2022-05-16 11:59:14

1.4K0

发布于 2022-05-16 11:59:14

文章被收录于专栏：软件测试技术软件测试技术

使用python操作kafka

安装 pip install kafka-python==2.0.2

kafka 的Producer

如果是kafka集群则bootstrap_servers可传入多个，需要使用逗号隔开。需要主要传入的值，必须转换为byte类型。

# TestKafkaProducer.py
from kafka import KafkaProducer

producer = KafkaProducer(bootstrap_servers=['12.23.34.56:9092'])

value = {"type": "test",
         "value": {"requestNo":"1","auditStatus":"2","failReason":"3"}}

bytesDict = bytes('{}'.format(value),'utf-8')

print(bytesDict)

topic='test_topic'

producer.send(topic, bytesDict)
producer.close()

kafka的Consumer

需要注意topic和bootstrap_servers地址 同上面一致。

# 安装 pip install  kafka-python==2.0.2
from kafka import KafkaConsumer
import time

topic='test_topic'

consumer = KafkaConsumer(topic, bootstrap_servers = ['12.23.34.56:9092'])

for m in consumer:
    print(m)
    print(m.topic)