對(duì)python操作kafka寫入json數(shù)據(jù)的簡(jiǎn)單demo分享

發(fā)布時(shí)間：2020-09-17 14:14:17 來(lái)源：腳本之家閱讀：213 作者：Liu-YanLin 欄目：開發(fā)技術(shù)

如下所示：

安裝kafka支持庫(kù)pip install kafka-python

from kafka import KafkaProducer
import json
 
'''
 生產(chǎn)者demo
 向test_lyl2主題中循環(huán)寫入10條json數(shù)據(jù)
 注意事項(xiàng)：要寫入json數(shù)據(jù)需加上value_serializer參數(shù)，如下代碼
'''
producer = KafkaProducer(
       value_serializer=lambda v: json.dumps(v).encode('utf-8'),
       bootstrap_servers=['192.168.12.101:6667','192.168.12.102:6667','192.168.12.103:6667']
       )
for i in range(10):
 data={
  "name":"李四",
  "age":23,
  "gender":"男",
  "id":i
 }
 producer.send('test_lyl2', data)
producer.close()

from kafka import KafkaConsumer
import json
 
'''
 消費(fèi)者demo
 消費(fèi)test_lyl2主題中的數(shù)據(jù)
 注意事項(xiàng)：如需以json格式讀取數(shù)據(jù)需加上value_deserializer參數(shù)
'''
 
 
consumer = KafkaConsumer('test_lyl2',group_id="lyl-gid1",
       bootstrap_servers=['192.168.12.101:6667','192.168.12.102:6667','192.168.12.103:6667'],
       auto_offset_reset='earliest',value_deserializer=json.loads
       )
for message in consumer:
 print(message.value)

以上這篇對(duì)python操作kafka寫入json數(shù)據(jù)的簡(jiǎn)單demo分享就是小編分享給大家的全部?jī)?nèi)容了，希望能給大家一個(gè)參考，也希望大家多多支持億速云。

向AI問(wèn)一下細(xì)節(jié)

對(duì)python操作kafka寫入json數(shù)據(jù)的簡(jiǎn)單demo分享

猜你喜歡

最新資訊

相關(guān)推薦

相關(guān)標(biāo)簽