在 Spring Boot 中配置 Kafka 消费者 = 失去对消费者行为的控制？

Question

我真的很喜欢使用 Spring Boot 中可用的所有配置来配置 Kafka 消费者：

ConsumerFactory() => KafkaListenerContainerFactory() => consume(V message)

但是，我似乎也失去了对 apache kafka 包附带的消费者行为的所有控制，例如在同步和异步提交之间来回切换，显式启动消费者并干净地关闭它。 使用 Spring kafka 接口，只需要实现一个方法，就可以开始获取消息了：

@KafkaListener(topics = "#{'${kafka.topics}'.split(',')}", containerFactory = "kafkaListenerContainerFactory")
public void consumeString(String message) {
    System.out.println("Consumed message: " + message);
}

在我看来，所有这些消耗都发生在 Spring Boot 自动提供的单独线程上...... 有人能告诉我如何在使用 Spring Boot 配置 KafkaConsumers（以及同样的原因 KafkaProducers）时仍然保持所有控制吗？ 谢谢！

Answer 1

在我看来，所有这些消耗都发生在 Spring Boot 自动提供的单独线程上

没错，spring 将 kafka 消费者包装在 listener 容器中。

您最多可以通过container properties 配置行为。

理论上，您可以保留kafka consumer factory 返回的最新值，但您需要记住，对于单个容器，可以多次创建消费者。无论如何，通过保留参考，您可以访问底层 kafka 消费者 反射。但这是一种技巧，您可能会影响容器的行为。