Kafka生产者flush和poll的区别

Question

我们有一个 Kafka 消费者，它将读取消息并执行此操作，然后使用以下脚本再次发布到 Kafka 主题

生产者配置：

{
  "bootstrap.servers": "localhost:9092"
}

我还没有配置任何其他配置像queue.buffering.max.messagesqueue.buffering.max.msbatch.num.messages

我假设这些都将是来自configuration的默认值

queue.buffering.max.messages : 100000
queue.buffering.max.ms : 0
batch.num.messages : 10000

我的理解：当内部队列达到 queue.buffering.max.ms 或 batch.num.messages 时，消息将在单独的线程中发布到 Kafka。在我的配置中 queue.buffering.max.ms 是 0，所以当我调用生产（）时，每条消息都会被发布。如果我错了，请纠正我。

我的制作人 sn-p：

def send(topic, message):
    p.produce(topic, json.dumps(message), callback=delivery_callback(err, msg))
    p.flush()

来自this post 我了解到，在每条消息后使用刷新，生产者将成为同步生产者。如果我使用上面的脚本，发布到 Kafka 大约需要 45 毫秒

如果我将 sn-p 更改为

def send(topic, message):
    p.produce(topic, json.dumps(message), callback=delivery_callback(err, msg))
    p.poll(0)

有没有什么性能会提高？你能澄清我的理解吗？

谢谢

Answer 1

flush() 和poll() 的区别在客户端的documentation 中有说明。

对于flush()，它声明：

等待生产者队列中的所有消息被传递。这是一个 调用 poll() 直到 len() 为零或 可选超时时间已过。

对于poll()：

轮询生产者以获取事件并调用相应的回调 （如果已注册）。

在send() 之后调用poll() 不会使生产者同步，因为刚刚发送的消息不太可能已经到达代理并且已经将传递报告发送回客户端。

相反，flush() 将阻塞，直到先前发送的消息已传递（或错误），从而有效地使生产者同步。