Python向kafka发消息

何艳娥

后端研发可以提供一个向kafka发消息的接口，用requests向接口post消息就行：

1. import requests
2. import json
3. import time
5. now = int(time.time())
6. n = 10
7. while n > 0:
8.     tt = now - n * 60
9.     data = {
10.         "queue": "alarm-dog-alarm-dog-test",
11.         "payload": "{"test":80,"notice_time":%d}" % tt
12.     }
13.     header = {"Content-Type": "application/json"}
14.     res = requests.post(url="http://10.90.100.130:8088/v1/kafka/send", headers=header, data=json.dumps(data))
15.     print(res.status_code)
16.     print(res.content)
17.     n -= 1

复制代码

如果没有提供接口，可以借助python-kafka库连接kafka，模拟生产者向kafka发消息：
同步发送消息：

1. from kafka import KafkaProducer
2. import json
4. # 创建一个KafkaProducer实例，指定Kafka服务器地址
5. producer = KafkaProducer(bootstrap_servers='http://10.90.100.130:8088')
6. # 要发送的消息内容
7. message = {'test': 80, 'notice_time': 5}
8. # 将消息转换为JSON字符串格式（也可以是其他格式，如纯文本）
9. message_json = json.dumps(message)
10. # 发送消息到指定的Kafka主题，这里主题名称是'my_topic'
11. producer.send('alarm-dog-alarm-dog-test', value=message_json.encode('utf - 8'))
12. # 确保所有消息都已发送
13. producer.flush()
14. # 关闭生产者连接
15. producer.close()

复制代码

异步发送消息

1. from kafka import KafkaProducer
2. import json
3. import time
5. # 创建一个KafkaProducer实例，设置异步发送和回调函数
6. producer = KafkaProducer(bootstrap_servers='http://10.90.100.130:8088',
7.           acks='all',
8.           retries=3,
9.           value_deliver_callback=lambda m: print(f"消息已发送到主题{m.topic()}，分区{m.partition()}"))
11. # 要发送的消息内容
12. message = {'test': 80, 'notice_time': 6}
13. message_json = json.dumps(message)
15. # 异步发送消息到'my_topic'主题
16. future = producer.send('alarm-dog-alarm-dog-test', value=message_json.encode('utf - 8'))
17. try:
18. record_metadata = future.get(timeout=10)
19. print(f"消息已发送到主题{record_metadata.topic()}，分区{record_metadata.partition()}，偏移量{record_metadata.offset()}")
20. except Exception as e:
21. print(f"发送消息时出错: {e}")
22. # 关闭生产者连接
23. producer.close()

复制代码

 

来源:https://www.cnblogs.com/ailiailan/p/18523653
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！