Kafka 入門（四）-- Python Kafka Client 性能測試

一、前言

　　由於工作原因使用到了 Kafka，而現有的代碼並不能滿足性能需求，所以需要開發高效讀寫 Kafka 的工具，本文是一個 Python Kafka Client 的性能測試記錄，通過本次測試，可以知道選用什么第三方庫的性能最高，選用什么編程模型開發出來的工具效率最高。

 

二、第三方庫性能測試

1.第三方庫

　　此次測試的是三個主要的 Python Kafka Client：pykafka、kafka-python 和 confluent-kafka，具體介紹見官網：

• pykafka：https://pypi.org/project/pykafka/
• kafka-python：https://pypi.org/project/kafka-python/
• confluent_kafka：https://pypi.org/project/confluent-kafka/

2.測試環境

       此次測試使用的 Python 版本是2.7，第三方庫的版本為：

• pykafka：2.8.0
• kafka-python：2.0.2
• confluent-kafka：1.5.0

       使用的數據總量有50萬，每條數據大小為2KB，總共為966MB。

3.測試過程

（1）Kafka Producer 測試

　　分別使用 pykafka、kafka-python 和 confluent-kafka 實例化一個 Kafka 的 Producer 對象，然后調用相應的 produce 方法將數據推送給 Kafka，數據總條數為50萬，比較三個庫所耗費的時間，並計算每秒鍾可以推送的數據條數和大小，比較得出性能最優的。

　　代碼示例（以 pykafka 為例）：

import sys
from datetime import datetime
from pykafka import KafkaClient


class KafkaProducerTool():
    def __init__(self, broker, topic):
        client = KafkaClient(hosts=broker)
        self.topic = client.topics[topic]
        self.producer = self.topic.get_producer()

    def send_msg(self, msg):
        self.producer.produce(msg)


if __name__ == '__main__':
    producer = KafkaProducerTool(broker, topic)
    print(datetime.now())
    for line in sys.stdin:
        producer.send_msg(line.strip())
    producer.producer.stop()
    print(datetime.now())

（2）Kafka Consumer 測試

　　分別使用 pykafka、kafka-python 和 confluent-kafka 實例化一個 Kafka 的 Consumer 對象，然后調用相應的 consume 方法從 Kafka 中消費數據，要消費下來的數據總條數為50萬，比較三個庫所耗費的時間，並計算每秒鍾可以消費的數據條數和大小，比較得出性能最優的。

　　代碼示例（以 pykafka 為例）：

from datetime import datetime
from pykafka import KafkaClient


class KafkaConsumerTool():
    def __init__(self, broker, topic):
        client = KafkaClient(hosts=broker)
        self.topic = client.topics[topic]
        self.consumer = self.topic.get_simple_consumer()

    def receive_msg(self):
        count = 0
        print(datetime.now())
        while True:
            msg = self.consumer.consume()
            if msg:
                count += 1
            if count == 500000:
                print(datetime.now())
                return


if __name__ == '__main__':
    consumer = KafkaConsumerTool(broker, topic)
    consumer.receive_msg()
    consumer.consumer.stop()

4.測試結果

• Kafka Producer 測試結果：

	總耗時/秒	每秒數據量/MB	每秒數據條數
confluent_kafka	35	27.90	14285.71
pykafka	50	19.53	10000
kafka-python	532	1.83	939.85

• Kafka Consumer 測試結果：

	總耗時/秒	每秒數據量/MB	每秒數據條數
confluent_kafka	39	25.04	12820.51
kafka-python	52	18.78	9615.38
pykafka	335	2.92	1492.54

5.測試結論

　　經過測試，在此次測試的三個庫中，生產消息的效率排名是：confluent-kafka > pykafka > kafka-python，消費消息的效率排名是：confluent-kafka > kafka-python > pykafka，由此可見 confluent-kafka 的性能是其中最優的，因而選用這個庫進行后續開發。

 

三、多線程模型性能測試

1.編程模型

　　經過前面的測試已經知道 confluent-kafka 這個庫的性能是很優秀的了，但如果還需要更高的效率，應該怎么辦呢？當單線程（或者單進程）不能滿足需求時，我們很容易想到使用多線程（或者多進程）來增加並發提高效率，考慮到線程的資源消耗比進程少，所以打算選用多線程來進行開發。那么多線程消費 Kafka 有什么實現方式呢？我想到的有兩種：

1. 一個線程實現一個 Kafka Consumer，最多可以有 n 個線程同時消費 Topic（其中 n 是該 Topic 下的分區數量）；
2. 多個線程共用一個 Kafka Consumer，此時也可以實例化多個 Consumer 同時消費。

　　　　

　　對比這兩種多線程模型：

• 模型1實現方便，可以保證每個分區有序消費，但 Partition 數量會限制消費能力；
• 模型2並發度高，可擴展能力強，消費能力不受 Partition 限制。

 2.測試過程

（1）多線程模型1

　　測試代碼：

import time
from threading import Thread
from datetime import datetime
from confluent_kafka import Consumer


class ChildThread(Thread):
    def __init__(self, name, broker, topic):
        Thread.__init__(self, name=name)
        self.con = KafkaConsumerTool(broker, topic)

    def run(self):
        self.con.receive_msg()


class KafkaConsumerTool:
    def __init__(self, broker, topic):
        config = {
            'bootstrap.servers': broker,
            'session.timeout.ms': 30000,
            'auto.offset.reset': 'earliest',
            'api.version.request': False,
            'broker.version.fallback': '2.6.0',
            'group.id': 'test'
        }
        self.consumer = Consumer(config)
        self.topic = topic

    def receive_msg(self):
        self.consumer.subscribe([self.topic])
        print(datetime.now())
        while True:
            msg = self.consumer.poll(timeout=30.0)
            print(msg)


if __name__ == '__main__':
    thread_num = 10
    threads = [ChildThread("thread_" + str(i + 1), broker, topic) for i in range(thread_num)]

    for i in range(thread_num):
        threads[i].setDaemon(True)
    for i in range(thread_num):
        threads[i].start()

　　因為我使用的 Topic 共有8個分區，所以我分別測試了線程數在5個、8個和10個時消費50萬數據所需要的時間，並計算每秒可消費的數據條數。

（2）多線程模型2

　　測試代碼：

import time
from datetime import datetime
from confluent_kafka import Consumer
from threadpool import ThreadPool, makeRequests


class KafkaConsumerTool:
    def __init__(self, broker, topic):
        config = {
            'bootstrap.servers': broker,
            'session.timeout.ms': 30000,
            'auto.offset.reset': 'earliest',
            'api.version.request': False,
            'broker.version.fallback': '2.6.0',
            'group.id': 'mini-spider'
        }
        self.consumer = Consumer(config)
        self.topic = topic

    def receive_msg(self, x):
        self.consumer.subscribe([self.topic])
        print(datetime.now())
        while True:
            msg = self.consumer.poll(timeout=30.0)
            print(msg)


if __name__ == '__main__':
    thread_num = 10
    consumer = KafkaConsumerTool(broker, topic)
    pool = ThreadPool(thread_num)
    for r in makeRequests(consumer.receive_msg, [i for i in range(thread_num)]):
        pool.putRequest(r)
    pool.wait()

　　主要使用 threadpool 這個第三方庫來實現線程池，此處當然也可以使用其他庫來實現，這里我分別測試了線程數量在5個和10個時消費50萬數據所需要的時間，並計算每秒可消費的數據條數。

3.測試結果

• 多線程模型1

總數據量/萬	線程數量	總耗時/秒	每秒數據條數
50	5	27	18518.51
50	8	24	20833.33
50	10	26	19230.76

• 多線程模型2

總數據量/萬	線程數量	總耗時/秒	每秒數據條數
50	5	17	29411.76
50	10	13	38461.53

4.測試結論

　　使用多線程可以有效提高 Kafka 的 Consumer 消費數據的效率，而選用線程池共用一個 KafkaConsumer 的消費方式的消費效率更高。