怎么在python3中连接kafka模块？针对这个问题，这篇文章详细介绍了相对应的分析和解答，希望可以帮助更多想解决这个问题的小伙伴找到更简单易行的方法。
创新互联建站不只是一家网站建设的网络公司；我们对营销、技术、服务都有自己独特见解，公司采取“创意+综合+营销”一体化的方式为您提供更专业的服务！我们经历的每一步也许不一定是最完美的，但每一步都有值得深思的意义。我们珍视每一份信任，关注我们的成都做网站、成都网站制作质量和服务品质，在得到用户满意的同时，也能得到同行业的专业认可，能够为行业创新发展助力。未来将继续专注于技术创新，服务升级，满足企业一站式
全网营销推广需求，让再小的
品牌网站设计也能产生价值！
1.1安装模块
pip install pykafka
1.2基本使用
# -* coding:utf8 *-
from pykafka import KafkaClient
host = 'IP:9092, IP:9092, IP:9092'
client = KafkaClient(hosts = host)
# 生产者
topicdocu = client.topics['my-topic']
producer = topicdocu.get_producer()
for i in range(100):
print i
producer.produce('test message ' + str(i ** 2))
producer.stop()1.3简单封装
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,datas):
"""
发送数据
:param datas:需要传入的可迭代对象
:return:
"""
c = self.Producer()
next(c)
for i in datas:
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模块问题的解答就分享到这里了，希望以上内容可以对大家有一定的帮助，如果你还有很多疑惑没有解开，可以关注创新互联行业资讯频道了解更多相关知识。
网站栏目：怎么在python3中连接kafka模块-创新互联
当前路径：
http://cdkjz.cn/article/ihoog.html
