python3连接kafka模块pykafka生产者简单封装代码

yipeiwu_com6年前Python基础

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模块pykafka生产者简单封装代码就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持【听图阁-专注于Python设计】。

相关文章

docker django无法访问redis容器的解决方法

docker django无法访问redis容器的解决方法

docker-compose.yal文件中: redis: image: redis container_name: xdemo.redis ports: - 6...

详解Python二维数组与三维数组切片的方法

如果对象是二维数组,则切片应当是x[:]的形式,里面有一个冒号,冒号之前和之后分别表示对象的第0个维度和第1个维度; 如果对象是三维数组,则切片应当是x[::],里面有两个冒号,分割出...

Python中Collections模块的Counter容器类使用教程

Python中Collections模块的Counter容器类使用教程

1.collections模块 collections模块自Python 2.4版本开始被引入,包含了dict、set、list、tuple以外的一些特殊的容器类型,分别是: Order...

python使用PyQt5的简单方法

python使用PyQt5的简单方法

一:安装PyQt5 pip install pyqt5 二:PyQt5简单使用 1:使用PyQt5创建一个简单窗口 import sys from PyQt5 import...

简单了解python模块概念

本文主要讲述的是Python中的模块的概念,具体如下。 模块是python组织代码的基本方式: python的脚本都是用扩展名为py的文本文件保存的。 一个脚本可以单独运行,也可以导入...