python3连接kafka模块pykafka生产者简单封装代码
1.1安装模块
pipinstallpykafka
1.2基本使用
#-*coding:utf8*-
frompykafkaimportKafkaClient
host='IP:9092,IP:9092,IP:9092'
client=KafkaClient(hosts=host)
#生产者
topicdocu=client.topics['my-topic']
producer=topicdocu.get_producer()
foriinrange(100):
printi
producer.produce('testmessage'+str(i**2))
producer.stop()
1.3简单封装
classKafkaProduct():
def__init__(self,hosts,topic):
"""
初始化实例
:paramhosts:连接地址
:paramtopic:
"""
self.__client=KafkaClient(hosts=hosts)
self.__topic=self.__client.topics[topic.encode()]
def__set_topic(self,topic):
self.__topic=self.__client.topics[topic.encode()]
defset_topic(self,topic):
"""
设置topic
:paramtopic:
:return:
"""
self.__set_topic(topic)
defget_topics(self):
"""
获取当前所有topic
:return:
"""
returnself.__client.topics
defget_topic(self):
"""
获取当前topic
:return:
"""
returnself.__topic
defProducer(self):
"""
生产者对象
:return:
"""
withself.__topic.get_producer(delivery_reports=True)asproducer:
next_data=''
whileTrue:
ifnext_data:
producer.produce(str(next_data).encode())
next_data=yieldTrue
defsend_data(self,datas):
"""
发送数据
:paramdatas:需要传入的可迭代对象
:return:
"""
c=self.Producer()
next(c)
foriindatas:
c.send(i)
if__name__=='__main__':
hosts="1.2.3.4:9999,2.3.4.5:9090"#连接hosts
topic="test_523"
K=KafkaProduct(hosts=hosts,topic=topic)#
#K.set_topic("test")#切换设置新的topic
K.get_topic()#获取当前设置的topic
#K.get_topics()#获取所有topic
data=range(10000)#要发送的可迭代对象
K.send_data(data)
以上这篇python3连接kafka模块pykafka生产者简单封装代码就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持毛票票。
声明：本文内容来源于网络，版权归原作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任。如果您发现有涉嫌版权的内容，欢迎发送邮件至：czq8825#qq.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立刻删除涉嫌侵权内容。