使用Python实现通过kafaka发送数据,具体代码如下:
from pykafka import KafkaClient
class KafkaProduct():
def __init__(self, hosts, topic):
"""
初始化实例
:param hosts: 连接地址
:param topic:主题
"""
self.client = KafkaClient(hosts=hosts)
self.topic = self.client.topics[topic.encode()]
def __set_topic(self, topic):
self.topic = self.client.topics[topic.encode()]
def set_topic(self, topic):
"""
设置topic
:param topic:
:return:
"""
self.set_topic(topic)
def get_topics(self):
"""
获取当前所有topic
:return:
"""
return self.client.topics
def get_topic(self):
"""
获取当前topic
:return:
"""
return self.topic
def Producer(self):
"""
生产者对象
:return:
"""
with self.topic.get_producer(delivery_reports=True) as producer:
next_data = ''
while True:
if next_data:
producer.produce(str(next_data).encode())
next_data = yield True
def send_data(self):
"""
发送数据
:param datas:需要传入的可迭代对象
:return:
"""
c = self.Producer()
next(c)
f = open("data.txt", "r", encoding='UTF-8')
datas = f.readlines()
for data in datas:
print(data)
c.send(data)
f.close()
if __name__ == '__main__':
hosts = "127.0.0.1:9092" # 连接hosts
topic = "test"
K = KafkaProduct(hosts=hosts, topic=topic) #
K.get_topic() # 获取当前设置的topic
K.get_topics() #获取所有topic
K.send_data()
本文转载自: https://blog.csdn.net/gcs12345/article/details/127618871
版权归原作者 gcs12345 所有, 如有侵权,请联系我们删除。
版权归原作者 gcs12345 所有, 如有侵权,请联系我们删除。