前言
由于工作原因使用到了 Kafka,而现有的代码并不能满足性能需求,所以需要开发高效读写 Kafka 的工具,本文是一个 Python Kafka Client 的性能测试记录,通过本次测试,可以知道选用什么第三方库的性能最高,选用什么编程模型开发出来的工具效率最高。
第三方库性能测试
1.第三方库
此次测试的是三个主要的 Python Kafka Client:pykafka、kafka-python 和 confluent-kafka,具体介绍见官网:
- pykafka:pykafka · PyPI
- kafka-python:kafka-python · PyPI
- confluent_kafka:confluent-kafka · PyPI
测试环境
此次测试使用的 Python 版本是2.7,第三方库的版本为:
pykafka:2.8.0
kafka-python:2.0.2
confluent-kafka:1.5.0
使用的数据总量有50万,每条数据大小为2KB,总共为966MB。
测试过程
(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()
测试结果
Kafka Producer 测试结果:
总耗时/秒每秒数据量/MB每秒数据条数confluent_kafka3527.9014285.71pykafka5019.5310000kafka-python5321.83939.85Kafka Consumer 测试结果:
总耗时/秒每秒数据量/MB每秒数据条数confluent_kafka3925.0412820.51kafka-python5218.789615.38pykafka3352.921492.54测试结论
经过测试,在此次测试的三个库中,生产消息的效率排名是:confluent-kafka > pykafka > kafka-python,消费消息的效率排名是:confluent-kafka > kafka-python > pykafka,由此可见 confluent-kafka 的性能是其中最优的,因而选用这个库进行后续开发。
多线程模型性能测试
编程模型
经过前面的测试已经知道 confluent-kafka 这个库的性能是很优秀的了,但如果还需要更高的效率,应该怎么办呢?当单线程(或者单进程)不能满足需求时,我们很容易想到使用多线程(或者多进程)来增加并发提高效率,考虑到线程的资源消耗比进程少,所以打算选用多线程来进行开发。那么多线程消费 Kafka 有什么实现方式呢?我想到的有两种:
- 一个线程实现一个 Kafka Consumer,最多可以有 n 个线程同时消费 Topic(其中 n 是该 Topic 下的分区数量);
- 多个线程共用一个 Kafka Consumer,此时也可以实例化多个 Consumer 同时消费。
对比这两种多线程模型:
- 模型1实现方便,可以保证每个分区有序消费,但 Partition 数量会限制消费能力;
- 模型2并发度高,可扩展能力强,消费能力不受 Partition 限制。
测试过程
(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 ge(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万数据所需要的时间,并计算每秒可消费的数据条数。
测试结果
多线程模型1
总数据量/万线程数量总耗时/秒每秒数据条数5052718518.515082420833.3350102619230.76多线程模型2
总数据量/万线程数量总耗时/秒每秒数据条数5051729411.7650101338461.53测试结论
使用多线程可以有效提高 Kafka 的 Consumer 消费数据的效率,而选用线程池共用一个 KafkaConsumer 的消费方式的消费效率更高。
版权归原作者 墨痕诉清风 所有, 如有侵权,请联系我们删除。