【kafka】生产者速度测试

时间：2022-09-11 15:03:44

非常有用的参考博客：http://blog.csdn.net/qq_33160722/article/details/52903380

pykafka文档：http://pykafka.readthedocs.io/en/latest/api/producer.html

起因：项目代码极慢，远远低于预期。后定位发现是kafka生产速度过慢导致。故检查原因。

先说结论：一定要在生产者退出前调用producer.stop()指令！！生产时用use_rdkafka=True参数。之前速度慢是由于没有调用该指令，保错后线程卡住导致的。

下面是添加了producer.stop()后的测试速度代码。

# coding=utf8

import os

from pykafka import KafkaClient

import json

import datetime

import traceback

broker_list = "xxxx:9092"

topic_name = "test"

client = KafkaClient(hosts=broker_list)

topic = client.topics[topic_name]

producer = topic.get_producer()   #多线程方式,异步  11s

#producer = topic.get_producer(use_rdkafka=True)   #使用rdkafka，异步  2s

#producer = topic.get_sync_producer() #同步

#producer = topic.get_producer(sync=True) #同步   13min

def send_data_kafka(data):

    try:

        msg = json.dumps(data)

        producer.produce(msg)

    except Exception:

        traceback.print_exc()

if __name__ == "__main__":

    for i in range(10000):

        d = {"ip": "127.0.0.1", "port": i, "msg": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"

                                                      "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}

        send_data_kafka(d)

    producer.stop()   # important !!!!!!!!!!!!!

另，附一个我自己测试时的小乌龙。

测试指定topic指定分区的日志总数的命令行指令。我一直加了--partitions参数，结果每次只能看到一个分区的数据。每次生产10000条数据，查看分区总数只有5000+条的增长，导致我以为有数据丢失。

后来发现，原来我指定的test话题有2个分区..... 不加--partitions就可以看到所有分区的数据了。

./kafka-run-class.sh kafka.tools.GetOffsetShell --topic test --time -1 --broker-list  xx.xx.xx.xx:9092 --partitions 0

推荐使用：

./kafka-run-class.sh kafka.tools.GetOffsetShell --topic test --time -1 --broker-list  xx.xx.xx.xx:9092

相关文章