python - PyKafka producer.get_delivery_report在block = false时抛出Queue.empty

我目前正在使用Python开发一个Kafka集成，我对Kafka和Python还不太熟悉，它们都来自PHP背景。
我已经设法让制作人工作，但由于等待来自卡夫卡的ack，它处理每条消息的速度不够快。
在GitHub页面（https://github.com/Parsely/pykafka）上，下面的示例应该异步处理消息，并且仍然允许传递报告：

>>> with topic.get_producer(delivery_reports=True) as producer:
...     count = 0
...     while True:
...         count += 1
...         producer.produce('test msg', partition_key='{}'.format(count))
...         if count % 10**5 == 0:  # adjust this or bring lots of RAM ;)
...             while True:
...                 try:
...                     msg, exc = producer.get_delivery_report(block=False)
...                     if exc is not None:
...                         print 'Failed to deliver msg {}: {}'.format(
...                             msg.partition_key, repr(exc))
...                     else:
...                         print 'Successfully delivered msg {}'.format(
...                         msg.partition_key)
...                 except Queue.Empty:
...                     break

我修改了示例，但是从测试中可以看出第一条消息发送成功，但是抛出了Queue.empty异常。
这是我修改过的代码：

from pykafka import KafkaClient
import Queue
import json

client = KafkaClient(hosts='1.1.1.1:9092')
topic = client.topics['test']


sync = False
# sync = True

if sync:
    with topic.get_sync_producer() as producer:
        count = 0
        while True:
            count += 1
            producer.produce('Test message ' + str(count))
            print 'Sent message ' + str(count)
else:
    with topic.get_producer(delivery_reports=True) as producer:
        count = 0
        while True:
            count += 1
            if count >= 100:
                print 'Processed 100 messages'
                break
            producer.produce('Test message ' + str(count))
            while True:
                try:
                    msg, exc = producer.get_delivery_report(block=False)
                    if exc is not None:
                        print 'Failed to deliver msg {}: {}'.format(msg.offset, repr(exc))
                    else:
                        print 'Successfully delivered msg {}'.format(msg.offset)
                except Queue.Empty:
                    print 'Queue.empty'
                    break

以及输出：

/Users/jim/Projects/kafka_test/env/bin/python /Users/jim/Projects/kafka_test/producer.py
Queue.empty
...
... x100
Processed 100 messages

通过检查我的消费者，我可以看到所有100条信息都已成功发送，但我无法从我的生产者这一点。
对于如何改进此实现，特别是如何在保持检查消息成功的能力的同时提高吞吐量，您有什么建议吗？

最佳答案

我发现了一个与此相关的GitHub问题：https://github.com/Parsely/pykafka/issues/291
我通过将最小排队消息降低到1来解决了这个问题。

with topic.get_sync_producer(min_queued_messages=1) as producer:
        count = 0
        while True:
            count += 1
            time_start = time.time()
            producer.produce('Test message ' + str(count))
            time_end = time.time()

            print 'Sent message %d, %ss duration' % (count, (time_end - time_start))

关于python - PyKafka producer.get_delivery_report在block = false时抛出Queue.empty，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/35298796/