您好,登录后才能下订单哦!
小编给大家分享一下python连接kafka的方法,希望大家阅读完这篇文章后大所收获,下面让我们一起去探讨方法吧!
1、kafka-python安装:
# PyPI安装 pip install kafka-python # conda安装 conda install -c conda-forge kafka-python # anaconda自带pip安装 /root/anaconda3/bin/pip install kafka-python
2、kafka-python生产者
producer.py
#!/usr/bin/env python # -*- coding: utf-8 -*- import datetime import json import time import uuid from kafka import KafkaProducer from kafka.errors import KafkaError producer = KafkaProducer(bootstrap_servers='100.69.222.221:9092,100.69.222.222:9092,100.69.222.223:9092') topic = 'test_20181105' def test(): print('begin') try: n = 0 while True: dic = {} dic['id'] = n n = n + 1 dic['myuuid'] = str(uuid.uuid4().hex) dic['time'] = datetime.datetime.now().strftime("%Y%m%d %H:%M:%S") producer.send(topic, json.dumps(dic).encode()) print("send:" + json.dumps(dic)) time.sleep(0.5) except KafkaError as e: print(e) finally: producer.close() print('done') if __name__ == '__main__': test()
服务器集群中配置好Kafka, 修改上面程序中的ip地址和端口号, 执行python脚本就可以成功将消息发送到 topic: test_20181105
send:{"id": 1411, "myuuid": "a25a3d0361f94d3b8fffd5967ab5df01", "time": "20181105 16:11:14"} send:{"id": 1412, "myuuid": "784efd5389564194941240dca66233b6", "time": "20181105 16:11:14"} send:{"id": 1413, "myuuid": "6a211195319e447aa559614662f70590", "time": "20181105 16:11:15"} send:{"id": 1414, "myuuid": "2cc45bd82baf4a1cb41ea4786e50a0df", "time": "20181105 16:11:15"} send:{"id": 1415, "myuuid": "b7dfed4919c74164b83cf3ec28e257b6", "time": "20181105 16:11:16"} send:{"id": 1416, "myuuid": "9218eceb17834c228f5ab01ca7595272", "time": "20181105 16:11:16"} send:{"id": 1417, "myuuid": "c2751c54c390453f9eedd417fb1e5a31", "time": "20181105 16:11:17"} send:{"id": 1418, "myuuid": "9bbc4ef2cfbb42148332eb979b1142cb", "time": "20181105 16:11:17"} send:{"id": 1419, "myuuid": "f4998a862494445c976137793b55ed73", "time": "20181105 16:11:18"}
3、kafka-python消费者
consumer.py
#!/bin/env python from kafka import KafkaConsumer # connect to Kafka server and pass the topic we want to consume consumer = KafkaConsumer('test_20181105',group_id = 'test_group2', bootstrap_servers='100.69.222.221:9092,100.69.222.222:9092,100.69.222.223:9092') try: for msg in consumer: print(msg) # print("%s:%d:%d: key=%s value=%s" % (msg.topic, msg.partition, msg.offset, msg.key, msg.value)) except KeyboardInterrupt as e: print(e)
同样修改上面的Ip地址和端口号,就可以接收 topic: test_20181105上的消息:
ConsumerRecord(topic='test_20181105', partition=1, offset=951, timestamp=1541405600340, timestamp_type=0, key=None, value=b'{"id": 1663, "myuuid": "0f744021b2d9468886908ee6685a0fdb", "time": "20181105 16:13:20"}', checksum=1357895145, serialized_key_size=-1, serialized_value_size=87) ConsumerRecord(topic='test_20181105', partition=0, offset=935, timestamp=1541405600841, timestamp_type=0, key=None, value=b'{"id": 1664, "myuuid": "9379f68f656644bdb2d30911f06240e4", "time": "20181105 16:13:20"}', checksum=-715594646, serialized_key_size=-1, serialized_value_size=87) ConsumerRecord(topic='test_20181105', partition=1, offset=952, timestamp=1541405601341, timestamp_type=0, key=None, value=b'{"id": 1665, "myuuid": "f4a5fa5b32cd4b7991612b626bea4b0e", "time": "20181105 16:13:21"}', checksum=-2068072013, serialized_key_size=-1, serialized_value_size=87)
可以通过设置不同的group_id 来实现消息队列或消息订阅:
如果所有的consumer都具有相同的group,这种情况和queue模式很像;消息将会在consumers之间负载均衡.
如果所有的consumer都具有不同的group,那这就是"发布-订阅";消息将会广播给所有的消费者.
看完了这篇文章,相信你对python连接kafka的方法有了一定的了解,想了解更多相关知识,欢迎关注亿速云行业资讯频道,感谢各位的阅读!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。