使用简单使用者在Apache Kafka中读取未处理的消息

Question

https://cwiki.apache.org/confluence/display/kafka/0.8.0+SimpleConsumer+example

使用SimpleConsumer来使用消息，但是在使用它时，我发现了一些突然的行为，如下所示：

使用者正在使用来自特定分区的消息。但问题是，当我的使用者运行并且我使用生产者将消息推送到主题时，它将使用来自该分区的消息。但是，如果我的消费者目前没有运行，我将一些消息推送到主题，然后再次启动消费者，它不会消费生产者推送的消息，但它再次准备消费现在将要推送的消息。我使用LatestTime（）代替EarliestTime（），因为我只想使用未处理的消息。

案例-1

使用者正在运行：

制作者将M1,M2,M3消息推送到主题1的分区1

生产者现在将m4,m5,m6 messgae推到主题1的分区1

现在调用consumer

结果：消费者不消费messgaes m4、m5、m6，但如果我要检查偏移量，那么它被设置为7。这意味着生产者在产生消息时已经将偏移量提前到7，结果消费者现在将使用偏移量为7的消息

请帮助理想情况下，当消费者再次出现时，它应该读取来自M4的消息。

Answer 1

你做错了。

首先，我不确定SimpleConsumer是您要找的东西。它迫使您自己管理偏移量（例如，它根本不向Zookeeper提交偏移量，并且每当您再次启动SimpleConsumer时，它将再次获取相同的消息）。SimpleConsumer不了解“已处理的消息”。它所能做的就是从某个偏移量开始获取，然后继续获取，直到你说“停止”。

无论如何，如果您打算自己提交处理过的偏移量，您应该使用earleSttime(auto.offset.reset=smallest配置项）。auto.offset.reset意味着如果使用错误的偏移量初始化消费者（如果我没记错的话，SimpleConsumer是使用-1偏移量初始化的，这显然是错误的），它将重置为可用的最小(earlesttime)或argest(latesttime)偏移量。

为了更清楚地说明，下面是一个示例：

您的案例-1:

创建一个使用者并将其指向主题1分区1。因为它最初是用错误的偏移量初始化的，所以它会要求代理提供一些适当的偏移量（这里是最小或最大偏移量重置）。如果您还没有生成任何消息，最小和最大的偏移量都为0，因此当您生成一些消息时，您的使用者将获取这些消息。

案例2:

您将生成N条消息（例如7）。然后启动SimpleConsumer。同样，它用错误的偏移量初始化，并向代理请求适当的偏移量。对于最小重置偏移量，它将为0，对于最大偏移量，它将为7。在您的示例中，您使用largestoffsets，您的使用者将使用offse7重新初始化并开始使用它。

一般来说，看看高级消费者，在大多数情况下，这是你要找的。这是链接