kafka异步产生丢失消息

Question

尝试按照网上的说明实现kafka异步生产。这是我的制作人的样子：

import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
public void asynSend(String topic, Integer partition, String message) {
    ProducerRecord<Object, Object> data = new ProducerRecord<>(topic, partition,null, message);
        producer.send(data, new DefaultProducerCallback());
    }

private static class DefaultProducerCallback implements Callback {
    @Override
    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
        if (e != null) {
            logger.error("Asynchronous produce failed");
        }
    }
}

我在这样的 for 循环中生成：

for (int i = 0; i < 5000; i++) {
    int partition = i % 2;
    FsProducerFactory.getInstance().asynSend(topic, partition,i + "th message to partition " + partition);
}

但是，某些消息可能会丢失。如下图，4508到4999的消息没有投递。

我发现原因可能是生产者进程关闭，缓存中所有当时未发送的消息都会丢失。 在 for 循环之后添加此行将解决此问题：

producer.flush();

但是，我不确定这是否是一个魅力解决方案，因为我注意到有人提到刷新会使异步发送以某种方式同步，任何人都可以解释或帮助我改进它吗？

Answer 1

在Kafka - The definitive Guide 书中，有一个异步生产者的示例，与您编写的代码完全相同。它使用send 和Callback。

在discussion 中写着：

在退出之前添加flush() 将使客户端等待任何未完成的消息传递给代理（这将在queue.buffering.max.ms 附近，加上延迟）。 如果您在每个 produce() 调用之后添加 flush()，那么您实际上是在实现一个同步生产者。

但如果你在for 循环之后进行，它不再是同步的，而是异步的。

您还可以将生产者配置中的acks 设置为all。这样，在主题的复制设置为大于 1 的情况下，您将有更多的保证来成功生成消息。