安装kafka-python:
C:\anaconda3\Scripts>pip install kafka-python
import datetime
import json
from kafka import KafkaProducer
from kafka import KafkaConsumer
from kafka.errors import KafkaError
'''
使用kafka-python的生产模块
'''
class Kafka_producer():
def __init__(self,bootstrapServers,kafkaTopic):
self.bootstrapServers = bootstrapServers
self.kafkaTopic = kafkaTopic
self.producer = KafkaProducer(bootstrap_servers=self.bootstrapServers)
def sendjsondata(self,params):
try:
parmas_message = json.dumps(params)
producer = self.producer
future = producer.send(self.kafkaTopic,parmas_message.encode('utf-8'))
producer.flush()
recordMetadata = future.get(timeout=10)
print(recordMetadata,datetime.datetime.now().strftime('%Y%m%d%H%M%S'))
except KafkaError as e:
print(e)
'''
使用Kafka-python的消费模块
'''
class Kafka_consumer():
def __init__(self,kafkaTopic,groupId):
self.kafkaTopic = kafkaTopic
self.bootstrapServers = bootstrapServers
self.groupId = groupId
self.consumer = KafkaConsumer(self.kafkaTopic,group_id=self.groupId,bootstrap_servers=self.bootstrapServers)
def consume_data(self):
try:
for message in self.consumer:
yield message
except BaseException as e:
print(e)
if __name__ == '__main__':
bootstrapServers = ['ip1:port1','ip2:port2','ip3:port3']
topicStr = '主题'
print('-' * 20)
print('生产者')
print('-' * 20)
producer = Kafka_producer(bootstrapServers,topicStr)
for id in range(5):
params = '{tst}:{null}---' + str(id)
producer.sendjsondata(params)
print('-' * 20)
print('消费者')
print('-' * 20)
groupId = 'group名称'
consumer = Kafka_consumer(bootstrapServers,topicStr,groupId)
message = consumer.consume_data()
for i in message:
print(i.value)
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。