【kafka】生產者速度測試

非常有用的參考博客：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