Apache Kafka 读取后手动删除队列答案

【问题标题】：Apache Kafka manual remove queue after readApache Kafka 读取后手动删除队列
【发布时间】：2018-09-26 03:34:14
【问题描述】：

我对 kafka 真的很陌生，我注意到在消费者阅读队列消息后，队列消息似乎已经消失了。如果这是主题队列的正常行为...有没有办法在阅读后不自动删除主题消息？

顺便说一句，这是我运行本地 kafka 服务器的命令：

.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties
.\bin\windows\kafka-server-start.bat .\config\server.properties

kafka 版本： kafka_2.12-1.0.0

我使用 kafka 存储将由我的邮件发送服务处理（读取）的所有消息。问题是......似乎在我的邮件服务器读取主题消息（邮件消息）之后，无论邮件发送过程是否成功，都永远消失了。

我想知道这对于 kafka 是否正常。
如果是，有没有办法在邮件发送成功后手动删除主题消息。

对不起，如果我的问题完全是愚蠢的。

【问题讨论】：

您有点困惑-在kafka中阅读后没有消息删除之类的东西（在特定时间后会删除消息，但您不应该这样做）。您应该正确管理您的阅读器偏移量 - 在您的情况下，可能只有在您成功完成处理后才提交它们。你可以在stackoverflow.com/questions/46546489/…看到更多信息

标签： spring spring-boot apache-kafka

【解决方案1】：

在您阅读消息后，您的消费者的偏移量会发生变化。

如果我正确理解您的问题，您可能需要从最早开始阅读。

看看 properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "最早");

【讨论】：

【解决方案2】：

不必抱歉。

在 kafka 中，消息（或记录）在使用后不会被删除。

您的消费者拥有名为“group.id”的属性，并且代理将偏移量关联到它。此偏移量也存储在另一个名为“_consumer”的 kafka 主题中。

如果您停止您的消费者并重新启动它（使用相同的 group.id 属性值），代理将向您发送更新的记录，该记录来自您的消费者在停止之前提交的最后一个偏移量。如果您使用另一个 group.id 启动另一个使用者，则您的 group.id 不存在偏移量，并且根据您决定使用以下使用者属性 properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); 的策略，您将重新使用所有主题。

如果不是这样，也许您的主题保留时间太短了？您可以使用以下命令进行检查：kafka-topics --zookeeper host_zoo:port_zoo --topic your_topic --describe

【讨论】：