java - Kafka - 生产者 - 句柄 "failed to send"

Question

我正在运行 0.8 Kafka，并使用提供的 Java API 构建生产者。
发送消息(或消息)的 API 函数返回 void。

有没有办法获取已发送消息的状态？是发送还是失败？

这对我们来说非常重要，因为我们正在从文件中读取消息，并且我们希望在发送完所有消息后删除该文件。但是，如果出现错误并且某些消息未发送并且我删除了该文件，则会导致非常重要的数据丢失。

Answer 1

您可以将您的生产者配置为等到它从 Kafka 集群获得 n acks (request.required.acks)，以便您在删除源文件之前有某种保证数据已正确提交。

如果真的您需要确保消息发送成功，您可能需要考虑让生产者同步的替代方案 (producer.type=sync)。这样，您将能够捕获阻塞调用抛出的任何异常并采取相应的行动。 send()抛出的异常是kafka.common.FailedToSendMessageException。

Kafka 的 Java API 并不理想，希望对您有所帮助。