pip install pykafka

# -* coding:utf8 *- 
from pykafka import KafkaClient 
host = 'IP:9092, IP:9092, IP:9092'
client = KafkaClient(hosts = host) 
# 生產(chǎn)者 
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()

class KafkaProduct():

 def __init__(self,hosts,topic):
  """
  初始化實(shí)例
  :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):
  """
  設(shè)置topic
  :param topic:
  :return:
  """
  self.__set_topic(topic)

 def get_topics(self):
  """
  獲取當(dāng)前所有topic
  :return:
  """
  return self.__client.topics

 def get_topic(self):
  """
  獲取當(dāng)前topic
  :return:
  """
  return self.__topic

 def Producer(self):
  """
  生產(chǎn)者對象
  :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):
  """
  發(fā)送數(shù)據(jù)
  :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") #切換設(shè)置新的topic
K.get_topic() #獲取當(dāng)前設(shè)置的topic
#K.get_topics() #獲取所有topic
data = range(10000) #要發(fā)送的可迭代對象
K.send_data(data)