Kakfa生产者消息传递与acks=all和flush

Question

使用“acks=all”配置创建Kafka制作人。

用上面的配置调用flush有什么意义吗？

在发送到代理之前，它是否会等待刷新被调用。

作为

acks=all这意味着领导者将等待完整的同步副本集来确认记录。这保证了只要至少有一个同步副本保持活动，记录就不会丢失。这是最强的可用保证。这相当于ack=-1设置。

Answer 1

在我回答这两个问题之前，让我先试着说出flush（）和acks之间的区别。

flush（）-这是一种在生产者中调用的方法，用于将消息从生产者端维护的缓冲区（可配置）推送到代理。您可以调用此方法或close（）将消息从生产者缓冲区发送到代理。如果生产者可用的缓冲区html" target="_blank">内存已满（如Manoj在回答中所述），则会自动调用该函数。

acks=ALL是代理的责任，也就是说，在消息按照生产者要求的设置同步复制到其他代理后，向生产者发送确认。您将使用此设置来调整您的消息传递语义学。在这种情况下，一旦消息复制到指定的同步副本，代理将向生产者发送确认，并说“我收到您的消息了”。

现在，关于你的问题，即是否有任何意义的调用flush与ack设置和生产者是否会等待flush被调用之前被发送到代理。

生产者的异步特性将确保生产者不会等待。但是，如果您显式调用flush（），或者如果它自己被调用，那么任何进一步的发送都将被阻止，直到生产者得到代理的确认。所以，这两者之间的关系非常微妙。

我希望这能有所帮助！

Answer 2

根据文件

齐平（）：

调用此方法可立即发送所有缓冲记录（即使linger_ms大于0），并在完成与这些记录相关的请求时阻止发送。flush（）的post条件是之前发送的任何记录都已完成（例如Future.is_done（）==True）。如果根据生产者的“acks”配置成功确认请求，或者请求导致错误，则认为请求已完成。

当一个线程被阻塞等待刷新调用完成时，其他线程可以继续发送消息；但是，不保证刷新调用开始后发送的消息是否完成。

flush（）仍将阻止客户端应用程序，直到发送所有消息，即使ack=0。唯一的一点是，它不会等待ack，该块只在缓冲区被发送出去之前。

flush（）with ack=all保证消息已发送，并已在集群上以所需的复制系数进行复制。

最后，回答您的问题：它会在被发送到代理之前等待flush被调用吗？

答：不一定，制作人一直在发信息 <罢工> 以间隔或 通过批处理大小（buffer.memory控制生产者可用存储器的总量以进行缓冲）。但是，flush（）总是好的，以确保您发送所有消息。

有关更多信息，请参阅此链接。