在 1.3.5 中停止 ConcurrentMessageListenerContainer

Question

我有一个问题，还有更多想了解的问题，以便我能以正确的方式实施它。要求是通过在容器上调用 stop 方法来停止 ConcurrentMessageListenerContainer，OOTB 为 KafkaMessageListenerContainer 迭代（基于定义的并发）并为每个消费者线程调用 stop。

仅供参考，我使用的是 1.3.5，由于 Spring Boot 1.5.*，我无法迁移到 2.*。

配置： 假设我有 5 个分区和并发定义为 5 的主题。使用 Batchlistener，因此每次轮询的批处理记录数 =100。

问题： 当我们在容器上调用 stop 时，它会出现在内部，它为每个 KafkaMessageListenerContainer 设置 running false 并在 listenerConsumer 上调用 wakeup。

setRunning(false);
this.listenerConsumer.consumer.wakeup();

在测试我观察到的内容时，通过在单独的线程中调用容器上的停止，它会执行以下操作：

1）它会停止 listenerConsumer，这肯定会起作用。调用 stop 后不再进行轮询。 2）似乎任何 listenerConsumer 已经轮询了 100 条记录，并且在处理过程中它在停止之前完成了执行。 #2 每个设计调用容器停止仅发送唤醒以停止下一次轮询？因为我在 KafkaMessageListenerContainer.run() 中看不到以下任何处理

catch (WakeupException e) {
     //Ignore, we're stopping
}

还有一件事，即使在 spring kafka 2.1 版本中，通过使用 ContainerStoppingBatchErrorHandler，它也会调用相同的容器停止，所以我想更多的是我理解如何处理这种情况..

总而言之，如果上面有更多详细信息，我想在从单独的线程调用 stop 时终止 listenerThread。我有手动偏移提交，所以重播批处理很好。

嗨，加里，

正如您建议让消费者感知侦听器一样，我的问题是特定于侦听器通过容器停止的。一旦容器调用停止，侦听器线程（无论是 BatchListner）都应该被中断执行。我知道侦听器已收到整个记录民意调查，问题不在于丢失偏移量，因为 ack 处于批处理级别。

Answer 1

不清楚您的问题是什么；当侦听器在处理完批处理后返回时，容器停止。

如果您想在记录级别管理偏移量，请不要使用BatchMessageListener；请改用记录级别的MessageListener。

对于前者，轮询返回的整批记录被视为一个工作单元；是否已提交整个批次的偏移量。

即使使用记录级监听器，在当前批次的记录发送到监听器之前，容器也不会停止；在这种情况下，您应该做什么取决于确认模式；使用手动确认，只需忽略停止后收到的那些记录。如果容器管理acks；为要丢弃的记录抛出异常（使用ackOnError=false - 默认为true）。

很遗憾，您无法迁移到更新的版本。

使用 2.1；有更多的灵活性。例如;提供了SeekToCurrentErrorHandler 和SeekToCurrentBatchErrorHandler。

SeekToCurrentErrorHandler 扩展了RemainingRecordsErrorHandler，这意味着轮询中的剩余记录被发送到错误处理程序而不是侦听器；当容器检测到有RemainingRecordsErrorHandler 时，侦听器将不会获取批处理中的剩余记录，并且处理程序可以决定做什么 - 停止容器、执行搜索等。

因此您不再需要在收到不良记录时停止容器。

这都是为了处理错误/由于某些数据错误而停止容器。

容器上目前没有stopNow() 选项 - 在处理当前记录后立即停止调用侦听器并丢弃所有剩余记录，以便重新发送它们。目前，您必须按照我上面的描述自己丢弃它们。

我们可以考虑添加这样一个选项，但它最早会是 2.2 的功能。

我们仅在 1.3.x 中提供错误修复，并且仅在没有解决方法时提供。

它是开源的，因此您始终可以自己派生它，并为框架做出贡献。