接收消息
你可以通过配置 MessageListenerContainer 并提供消息监听器来接收消息，也可以使用 @KafkaListener 注解来接收消息。
章节摘要
消息监听器
当你使用消息监听器容器时，必须提供一个监听器来接收数据。目前有八个受支持的消息监听器 接口。以下清单展示了这些接口：
消息监听器容器
提供了两种 MessageListenerContainer 实现：
手动提交偏移量
通常，在使用 AckMode.MANUAL 或 AckMode.MANUAL_IMMEDIATE 时，确认必须按顺序 进行，因为 Kafka 不为每条记录维护状态，而只为每个 group/partition 维护一个已提交的 偏移量。从版本 2.8 开始，你现在可以设置容器属性 asyncAcks，这使得为 poll 返回的 记录所做的确认可以按任意顺序进行确认。监听器容器会延迟乱序提交，直到收到缺失的 确认为止。Consumer 将被暂停（不投递新记录），直到上一次 poll 的所有偏移量都已提交。
异步 @KafkaListener 返回类型
从版本 3.2 开始，@KafkaListener（以及 @KafkaHandler）方法可以指定为异步返回类型， 从而让回复被异步发送。返回类型包括 CompletableFuture<?>、Mono<?> 和 Kotlin suspend 函数。
@KafkaListener 注解
@KafkaListener 注解用于将一个 bean 方法指定为监听器容器的监听器。该 bean 被包装在一个 MessagingMessageListenerAdapter 中，后者配置了各种特性，例如转换器，以便在必要时将数据 转换为与方法参数匹配。
获取 Consumer group.id
当在多个容器中运行相同的监听器代码时，能够确定一条记录来自哪个容器 （以其 group.id consumer 属性标识）可能会很有用。
容器线程命名
TaskExecutor 用于调用 consumer 和监听器。你可以通过设置容器 ContainerProperties 的 consumerExecutor 属性来提供自定义执行器。使用池化执行器时，请确保有足够的线程来处理 所有使用它们的容器之间的并发。使用 ConcurrentMessageListenerContainer 时， 执行器中的一个线程用于每个 consumer（concurrency）。
@KafkaListener 用作元注解
从版本 2.2 开始，你现在可以将 @KafkaListener 用作元注解。以下示例展示了 如何这样做：
类上的 @KafkaListener
当你在类级别使用 @KafkaListener 时，必须在方法级别指定 @KafkaHandler。如果该类或 其子类的任何方法上都没有 @KafkaHandler，框架将拒绝这样的配置。@KafkaHandler 注解 用于明确而简洁地表达方法的用途。否则，在没有额外限制的情况下，很难对这个方法或其他 方法做出决定。
@KafkaListener 属性修改
从版本 2.7.2 开始，你现在可以在容器创建之前以编程方式修改注解属性。为此，向应用 上下文添加一个或多个 KafkaListenerAnnotationBeanPostProcessor.AnnotationEnhancer。 AnnotationEnhancer 是一个 BiFunction<Map<String, Object>, AnnotatedElement, Map<String, Object>，并且必须返回一个属性 map。属性值可以包含 SpEL 和/或属性占位符； enhancer 在执行任何解析之前被调用。如果存在多个 enhancer，并且它们实现了 Ordered， 则将按顺序调用它们。
@KafkaListener 生命周期管理
为 @KafkaListener 注解创建的监听器容器不是应用上下文中的 bean。相反，它们被注册到一个 类型为 KafkaListenerEndpointRegistry 的基础架构 bean 中。该 bean 由框架自动声明并管理 这些容器的生命周期；它会自动启动任何将 autoStartup 设置为 true 的容器。由所有容器工厂 创建的所有容器必须处于相同的 phase。有关更多信息，请参阅监听器容器自动启动。你可以使用 registry 以编程方式管理生命周期。启动或停止 registry 将启动或停止所有已注册的容器。 或者，你可以使用其 id 属性获得对单个容器的引用。你可以在注解上设置 autoStartup，它会 覆盖配置到容器工厂中的默认设置。你可以从应用上下文（例如通过自动装配）获得对该 bean 的 引用，以管理其已注册的容器。以下示例展示了如何这样做：
@KafkaListener @Payload 校验
从版本 2.2 开始，添加 Validator 来校验 @KafkaListener @Payload 参数变得更加容易。 以前，你必须配置一个自定义的 DefaultMessageHandlerMethodFactory 并将其添加到 registrar。 现在，你可以将 validator 添加到 registrar 本身。以下代码展示了如何这样做：
再均衡监听器
ContainerProperties 有一个名为 consumerRebalanceListener 的属性，它接收 Kafka client 的 ConsumerRebalanceListener 接口的一个实现。如果不提供此属性，容器会配置一个日志监听器， 以 INFO 级别记录再均衡事件。框架还添加了一个子接口 ConsumerAwareRebalanceListener。 以下清单展示了 ConsumerAwareRebalanceListener 接口定义：
强制 Consumer 再均衡
Kafka clients 现在支持一个触发强制再均衡的选项。从版本 3.1.2 开始，Spring for Apache Kafka 提供了一个通过消息监听器容器在 Kafka consumer 上调用此 API 的选项。调用此 API 时，它只是提醒 Kafka consumer 触发一次强制再均衡；实际的再均衡只会在下一次 poll() 操作时发生。如果已经有一个再均衡正在进行中，调用强制再均衡是一个 NO-OP。 调用者必须等待当前再均衡完成后才能再调用另一个。有关详情，请参阅 enforceRebalance 的 javadoc。
使用 @SendTo 转发监听器结果
从版本 2.0 开始，如果你还用 @SendTo 注解标注了 @KafkaListener，并且方法调用返回了结果， 则该结果会被转发到 @SendTo 所指定的主题。
过滤消息
在某些场景下，例如再均衡，一条已经处理过的消息可能会被重新投递。框架无法知道这样的 消息是否已被处理过。那是应用层的功能。这被称为幂等接收器（Idempotent Receiver） 模式，Spring Integration 提供了它的一个实现。
重试投递
请参阅“处理异常”中的 DefaultErrorHandler。
按顺序启动 @KafkaListener
一个常见用例是：在另一个监听器消费完某个主题中的所有记录之后再启动一个监听器。例如， 你可能想在处理来自其他主题的记录之前，将一个或多个压缩主题的内容加载到内存中。 从版本 2.7.3 开始，引入了一个新组件 ContainerGroupSequencer。它使用 @KafkaListener 的 containerGroup 属性将容器分组在一起，并在当前组中的所有容器都转为空闲时启动下一组中的 容器。
使用 KafkaTemplate 接收
本节介绍如何使用 KafkaTemplate 接收消息。