对python操作kafka写入json数据的简单demo分享

yipeiwu_com6年前Python基础

如下所示:

安装kafka支持库pip install kafka-python
from kafka import KafkaProducer
import json
 
'''
 生产者demo
 向test_lyl2主题中循环写入10条json数据
 注意事项:要写入json数据需加上value_serializer参数,如下代码
'''
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
 
'''
 消费者demo
 消费test_lyl2主题中的数据
 注意事项:如需以json格式读取数据需加上value_deserializer参数
'''
 
 
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)

以上这篇对python操作kafka写入json数据的简单demo分享就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持【听图阁-专注于Python设计】。

相关文章

python内置函数sorted()用法深入分析

本文实例讲述了python内置函数sorted()用法。分享给大家供大家参考,具体如下: 列表对象提供了sort()方法支持原地排序,而内置函数sorted()不支持原地操作只是返回新的...

python数字图像处理之高级滤波代码详解

python数字图像处理之高级滤波代码详解

本文提供许多的滤波方法,这些方法放在filters.rank子模块内。 这些方法需要用户自己设定滤波器的形状和大小,因此需要导入morphology模块来设定。 1、autolevel...

Python中防止sql注入的方法详解

前言 大家应该都知道现在web漏洞之首莫过于sql了,不管使用哪种语言进行web后端开发,只要使用了关系型数据库,可能都会遇到sql注入攻击问题。那么在Python web开发的过程中s...

Python多叉树的构造及取出节点数据(treelib)的方法

项目: 基于Pymysql的专家随机抽取系统 引入库函数: >>> import treelib >>> from treelib import...

python设置环境变量的原因和方法

python设置环境变量的原因和方法

相信很多初学python的小伙伴都会遇到这样的坑:环境变量配置不好,无法正常启动python。那么环境变量究竟是个什么东西呢?为什么要设置它?下面我们来说一说。 1、什么是环境变量 引用...