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)者對象
        :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)

            
          

1.4引用來源

博客園:Python測試Kafka集群(pykafka)

知乎:使用生成器把Kafka寫入效率提高1000倍