怎么在python3中連接kafka模塊-創(chuàng)新互聯(lián)

怎么在python3中連接kafka模塊？針對(duì)這個(gè)問題，這篇文章詳細(xì)介紹了相對(duì)應(yīng)的分析和解答，希望可以幫助更多想解決這個(gè)問題的小伙伴找到更簡單易行的方法。

創(chuàng)新互聯(lián)建站不只是一家網(wǎng)站建設(shè)的網(wǎng)絡(luò)公司；我們對(duì)營銷、技術(shù)、服務(wù)都有自己獨(dú)特見解，公司采取“創(chuàng)意+綜合+營銷”一體化的方式為您提供更專業(yè)的服務(wù)！我們經(jīng)歷的每一步也許不一定是最完美的，但每一步都有值得深思的意義。我們珍視每一份信任，關(guān)注我們的成都做網(wǎng)站、成都網(wǎng)站制作質(zhì)量和服務(wù)品質(zhì)，在得到用戶滿意的同時(shí)，也能得到同行業(yè)的專業(yè)認(rèn)可，能夠?yàn)樾袠I(yè)創(chuàng)新發(fā)展助力。未來將繼續(xù)專注于技術(shù)創(chuàng)新，服務(wù)升級(jí)，滿足企業(yè)一站式全網(wǎng)營銷推廣需求，讓再小的品牌網(wǎng)站設(shè)計(jì)也能產(chǎn)生價(jià)值！

1.1安裝模塊

pip install pykafka

1.2基本使用

# -* 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()

1.3簡單封裝

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)者對(duì)象
  :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:需要傳入的可迭代對(duì)象
  :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ā)送的可迭代對(duì)象
K.send_data(data)

關(guān)于怎么在python3中連接kafka模塊問題的解答就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道了解更多相關(guān)知識(shí)。

當(dāng)前文章：怎么在python3中連接kafka模塊-創(chuàng)新互聯(lián)
分享地址：http://www.bm7419.com/article6/ihoog.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供電子商務(wù)、動(dòng)態(tài)網(wǎng)站、標(biāo)簽優(yōu)化、外貿(mào)網(wǎng)站建設(shè)、定制網(wǎng)站、面包屑導(dǎo)航

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請(qǐng)聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源：創(chuàng)新互聯(lián)

猜你還喜歡下面的內(nèi)容