Tomcat 中的 Kafka 消费者关闭答案

【问题标题】：Kafka Consumer Shutdown in TomcatTomcat 中的 Kafka 消费者关闭
【发布时间】：2018-08-15 19:54:39
【问题描述】：

我们正在 tomcat 服务器中部署 kafka 消费者。消费者是使用 spring-kafka 2.1.7 构建的。每个 tc 容器可以有多个消费者（使用 ConcurrentKafkaListenerContainerFactory）属于同一个消费者组。作为一般模式消费者在我的用例中，从一个主题读取并以事务方式生成另一个主题。Tc 服务器由通常的启动和关闭 shell 脚本启动和停止。为了优雅地关闭消费者，应该通过从内部调用 container.stop 来停止容器关机钩子？

【问题讨论】：

标签： spring-kafka

【解决方案1】：

您的担忧不清楚。 ConcurrentKafkaListenerContainerFactory 是一个 Spring bean，因此它的生命周期由 Spring ApplicationContext 控制。如果这个被正确部署到 Tomcat 中，default 会在那里注册一个适当的关闭挂钩：

Spring 基于 Web 的 ApplicationContext 实现已经有代码可以在相关 Web 应用程序关闭时优雅地关闭 Spring IoC 容器。

【讨论】：

谢谢阿特姆。让我重新措辞一下。如果我有@KafkaListener 并且tomcat 已关闭，是否有可能发生消息丢失或重复。我让容器使用 AckMode.RECORD 管理事务。考虑从源主题读取消息并写入目标主题但 tc 容器在偏移主题上发生提交之前关闭的场景。一个不相关的问题 - 如果我使用 Ackmode.BATCH 或COUNT 。例如，如果批处理失败，是否可以从失败的记录重试，或者这会导致重复记录
抱歉，这看起来与我们最初的问题无关。如果你从消费者到生产者使用KafkaTransactionManager，你不会失去任何东西。这正是卡夫卡的一次：confluent.io/blog/…。偏移量的提交被发送到生产者以应用于 TX 提交。