kafka-python开发kafka生产者和消费者
2024-09-08 04:04:38
1、安装kafka-python
执行命令
pip install kafka-python
kafka-python 1.4.6
2、编写python kafka 生产者消费者代码
# test.py import sys
import time
import json from kafka import KafkaProducer
from kafka import KafkaConsumer
from kafka.errors import KafkaError KAFAKA_HOST = "127.0.0.1"
KAFAKA_PORT = 9092
KAFAKA_TOPIC = "test123" class Kafka_producer():
'''''
生产模块:根据不同的key,区分消息
''' def __init__(self, kafkahost,kafkaport, kafkatopic, key):
self.kafkaHost = kafkahost
self.kafkaPort = kafkaport
self.kafkatopic = kafkatopic
self.key = key
print("producer:h,p,t,k",kafkahost,kafkaport,kafkatopic,key)
bootstrap_servers = '{kafka_host}:{kafka_port}'.format(
kafka_host=self.kafkaHost,
kafka_port=self.kafkaPort
)
print("boot svr:",bootstrap_servers)
self.producer = KafkaProducer(bootstrap_servers = bootstrap_servers
) def sendjsondata(self, params):
try:
parmas_message = json.dumps(params,ensure_ascii=False)
producer = self.producer
print(parmas_message)
v = parmas_message.encode('utf-8')
k = key.encode('utf-8')
print("send msg:(k,v)",k,v)
producer.send(self.kafkatopic, key=k, value= v)
producer.flush()
except KafkaError as e:
print (e) class Kafka_consumer():
'''''
消费模块: 通过不同groupid消费topic里面的消息
''' def __init__(self, kafkahost, kafkaport, kafkatopic, groupid):
self.kafkaHost = kafkahost
self.kafkaPort = kafkaport
self.kafkatopic = kafkatopic
self.groupid = groupid
self.key = key
self.consumer = KafkaConsumer(self.kafkatopic, group_id = self.groupid,
bootstrap_servers = '{kafka_host}:{kafka_port}'.format(
kafka_host=self.kafkaHost,
kafka_port=self.kafkaPort )
) def consume_data(self):
try:
for message in self.consumer:
yield message
print("")
print(message)
except KeyboardInterrupt as e:
print (e) def main(xtype, group, key):
'''''
测试consumer和producer
'''
if xtype == "p":
# 生产模块
producer = Kafka_producer(KAFAKA_HOST, KAFAKA_PORT, KAFAKA_TOPIC, key)
print ("===========> producer:", producer)
for _id in range(100):
# params = '{"msg" : "%s"}' % str(_id)
params=[{"msg0" :_id},{"msg1" :_id}]
producer.sendjsondata(params)
time.sleep(1) if xtype == 'c':
# 消费模块
consumer = Kafka_consumer(KAFAKA_HOST, KAFAKA_PORT, KAFAKA_TOPIC, group)
print ("===========> consumer:", consumer)
message = consumer.consume_data()
print('')
print(message)
for msg in message:
print ('msg---------------->k,v', msg.key,msg.value)
print ('offset---------------->', msg.offset) if __name__ == '__main__':
xtype = sys.argv[1]
group = sys.argv[2]
key = sys.argv[3]
main(xtype, group, key)
3、启动kafka服务
打开终端 输入命令:
kafka-server-start /usr/local/etc/kafka/server.properties
4、新开一终端创建生产者
切换到程序路径执行如下指令
python test.py p g k
5、新开一终端创建消费者
切换到程序路径执行如下指令
python test.py c g k
至此已经完成kafka 的消息收发了。
最新文章
- Visual Studio 生成事件命令
- sql语句修改列
- html5 弹框 可用于安卓手机弹出输入框
- javascript的执行和预解析
- vue学习笔记之v-if
- FreeMarker 一二事 - 静态模板结合spring展示
- poj 1986
- OC-KVO简介
- mysql数据库优化日志(更)-howyue
- Delphi XE7中新并行库
- Java笔记:Java集合概述和Set集合
- 标准模式 怪异模式 盒模型 doctype
- 直播-rtmp学习
- 海量数据挖掘MMDS week4: 推荐系统Recommendation System
- jdk12+tomcat9 配置
- 【Go】深入剖析slice和array
- iFrame跨域解决办法
- 单例&;多线程
- [Command] lrzsz - 文件传输工具包
- Tomcat配置JNDI数据源的三种方式-转-http://blog.51cto.com/xficc/1564691