消息监听器容器

Doubao 中英对照 Message Listener Containers

提供了两种 MessageListenerContainer 实现：

• KafkaMessageListenerContainer
• ConcurrentMessageListenerContainer

KafkaMessageListenerContainer 在单个线程上接收来自所有主题或分区的所有消息。ConcurrentMessageListenerContainer 委托给一个或多个 KafkaMessageListenerContainer 实例，以提供多线程消费。

从版本 2.2.7 开始，你可以向监听器容器添加一个 RecordInterceptor；它会在调用监听器之前被调用，允许检查或修改记录。如果拦截器返回 null，则不会调用监听器。从版本 2.7 开始，它有了附加方法，这些方法在监听器退出后（正常退出或抛出异常）被调用。此外，从版本 2.7 开始，现在有一个 BatchInterceptor，为批监听器提供类似的功能。另外，ConsumerAwareRecordInterceptor（和 BatchInterceptor）提供了对 Consumer<?, ?> 的访问。例如，这可以用于在拦截器中访问 consumer 指标。

important

你不应该在这些拦截器中执行任何会影响 consumer 位置和/或已提交偏移量的方法；这些信息需要由容器来管理。

important

如果拦截器修改了记录（通过创建一个新记录），则 topic、partition 和 offset 必须保持不变，以避免出现意外的副作用，例如记录丢失。

CompositeRecordInterceptor 和 CompositeBatchInterceptor 可用于调用多个拦截器。

从版本 4.0 开始，AbstractKafkaListenerContainerFactory 和 AbstractMessageListenerContainer 将 getRecordInterceptor() 和 getBatchInterceptor() 暴露为公开方法。如果返回的拦截器是 CompositeRecordInterceptor 或 CompositeBatchInterceptor 的实例，则即使在扩展 AbstractMessageListenerContainer 的容器实例已被创建、并且已经配置了 RecordInterceptor 或 BatchInterceptor 之后，也可以向其添加额外的 RecordInterceptor 或 BatchInterceptor 实例。以下示例展示了如何这样做：

public void configureRecordInterceptor(AbstractKafkaListenerContainerFactory<Integer, String> containerFactory) {
    CompositeRecordInterceptor compositeInterceptor;

    RecordInterceptor<Integer, String> previousInterceptor = containerFactory.getRecordInterceptor();
    if (previousInterceptor instanceof CompositeRecordInterceptor interceptor) {
        compositeInterceptor = interceptor;
    } else {
        compositeInterceptor = new CompositeRecordInterceptor<>();
        containerFactory.setRecordInterceptor(compositeInterceptor);
        if (previousInterceptor != null) {
            compositeInterceptor.addRecordInterceptor(previousInterceptor);
        }
    }

    RecordInterceptor<Integer, String> recordInterceptor1 = new RecordInterceptor() {...};
    RecordInterceptor<Integer, String> recordInterceptor2 = new RecordInterceptor() {...};

    compositeInterceptor.addRecordInterceptor(recordInterceptor1);
    compositeInterceptor.addRecordInterceptor(recordInterceptor2);
}

默认情况下，从版本 2.8 开始，在使用事务时，拦截器在事务开始之前被调用。你可以将监听器容器的 interceptBeforeTx 属性设置为 false，改为在事务开始之后调用拦截器。从版本 2.9 开始，这将适用于任何事务管理器，而不仅仅是 KafkaAwareTransactionManager。例如，这允许拦截器参与由容器启动的 JDBC 事务。

从版本 2.3.8、2.4.6 开始，当并发大于一时，ConcurrentMessageListenerContainer 现在支持静态成员资格。group.instance.id 会加上后缀 -n，其中 n 从 1 开始。这一点，配合增大的 session.timeout.ms，可用于减少再均衡事件，例如在重启应用实例时。

使用 KafkaMessageListenerContainer

提供了以下构造函数：

public KafkaMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
                    ContainerProperties containerProperties)

它接收一个 ConsumerFactory，以及关于主题和分区的信息，还有其他配置，都封装在一个 ContainerProperties 对象中。ContainerProperties 有以下构造函数：

public ContainerProperties(TopicPartitionOffset... topicPartitions)

public ContainerProperties(String... topics)

public ContainerProperties(Pattern topicPattern)

第一个构造函数接收一个 TopicPartitionOffset 参数数组，以显式指示容器使用哪些分区（使用 consumer 的 assign() 方法），并带有可选的初始偏移量。正值默认表示绝对偏移量。负值默认表示相对于分区内当前最后偏移量的相对值。还提供了一个接收附加 boolean 参数的 TopicPartitionOffset 构造函数。如果为 true，则初始偏移量（正或负）是相对于该 consumer 的当前位置。偏移量在容器启动时应用。第二个构造函数接收一个主题数组，Kafka 会基于 group.id 属性分配分区——将分区分布到整个组。第三个使用正则 Pattern 来选择主题。

要将 MessageListener 分配给容器，你可以在创建 Container 时使用 ContainerProps.setMessageListener 方法。以下示例展示了如何这样做：

ContainerProperties containerProps = new ContainerProperties("topic1", "topic2");
containerProps.setMessageListener(new MessageListener<Integer, String>() {
    ...
});
DefaultKafkaConsumerFactory<Integer, String> cf =
                        new DefaultKafkaConsumerFactory<>(consumerProps());
KafkaMessageListenerContainer<Integer, String> container =
                        new KafkaMessageListenerContainer<>(cf, containerProps);
return container;

请注意，在创建 DefaultKafkaConsumerFactory 时，使用上面那样只接收属性的构造函数意味着 key 和 value 的 Deserializer 类是从配置中获取的。或者，可以将 Deserializer 实例传递给 DefaultKafkaConsumerFactory 构造函数用于 key 和/或 value，在这种情况下，所有 Consumer 共享相同的实例。另一个选择是提供 Supplier<Deserializer>（从版本 2.3 开始），它们将被用于为每个 Consumer 获取单独的 Deserializer 实例：

DefaultKafkaConsumerFactory<Integer, CustomValue> cf =
                        new DefaultKafkaConsumerFactory<>(consumerProps(), null, () -> new CustomValueDeserializer());
KafkaMessageListenerContainer<Integer, String> container =
                        new KafkaMessageListenerContainer<>(cf, containerProps);
return container;

请参阅 ContainerProperties 的 Javadoc，了解关于可以设置的各种属性的更多信息。

从版本 2.1.1 开始，提供了一个名为 logContainerConfig 的新属性。当为 true 且启用了 INFO 日志记录时，每个监听器容器都会写入一条日志消息，总结其配置属性。

默认情况下，主题偏移量提交的日志记录在 DEBUG 日志级别进行。从版本 2.1.2 开始，ContainerProperties 中一个名为 commitLogLevel 的属性让你可以指定这些消息的日志级别。例如，要将日志级别更改为 INFO，你可以使用 containerProperties.setCommitLogLevel(LogIfLevelEnabled.Level.INFO);。

从版本 2.2 开始，添加了一个名为 missingTopicsFatal 的新容器属性（自 2.3.4 起默认为 false）。如果任何已配置的主题在 broker 上不存在，这会阻止容器启动。如果容器被配置为监听主题模式（正则），则不适用。以前，容器线程在 consumer.poll() 方法内循环等待主题出现，同时记录大量消息。除了日志之外，没有迹象表明出了问题。

从版本 2.8 开始，引入了一个新的容器属性 authExceptionRetryInterval。这使得容器在从 KafkaConsumer 收到任何 AuthenticationException 或 AuthorizationException 之后重试获取消息。例如，当配置的用户被拒绝读取某个主题的权限或凭据不正确时，就会发生这种情况。定义 authExceptionRetryInterval 可让容器在获得适当权限后恢复。

备注

默认情况下，不配置间隔——认证和授权错误被视为致命错误，会导致容器停止。

从版本 2.8 开始，在创建 consumer 工厂时，如果你以对象形式提供反序列化器（在构造函数中或通过 setter），该工厂将调用 configure() 方法，用配置属性来配置它们。

使用 ConcurrentMessageListenerContainer

唯一的构造函数与 KafkaListenerContainer 构造函数类似。以下清单展示了该构造函数的签名：

public ConcurrentMessageListenerContainer(ConsumerFactory<K, V> consumerFactory,
                            ContainerProperties containerProperties)

它还有一个 concurrency 属性。例如，container.setConcurrency(3) 会创建三个 KafkaMessageListenerContainer 实例。

如果容器属性配置为主题（或主题模式），Kafka 会使用其组管理功能将分区分布到各个 consumer。

important

当监听多个主题时，默认的分区分布可能与你预期的不同。例如，如果你有三个主题、每个主题五个分区，并且你想使用 concurrency=15，你会看到只有五个活跃的 consumer，每个从每个主题分配一个分区，而其余 10 个 consumer 处于空闲状态。这是因为默认的 Kafka ConsumerPartitionAssignor 是 RangeAssignor（请参阅其 Javadoc）。对于这种场景，你可能想考虑改用 RoundRobinAssignor，它将分区分布到所有 consumer 上。这样，每个 consumer 被分配一个主题或分区。要更改 ConsumerPartitionAssignor，你可以在提供给 DefaultKafkaConsumerFactory 的属性中设置 partition.assignment.strategy consumer 属性（ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG）。

使用 Spring Boot 时，你可以如下设置该策略：

spring.kafka.consumer.properties.partition.assignment.strategy=\
org.apache.kafka.clients.consumer.RoundRobinAssignor

:::

当容器属性配置了 TopicPartitionOffset 时，ConcurrentMessageListenerContainer 会将这些 TopicPartitionOffset 实例分布到各个委托的 KafkaMessageListenerContainer 实例上。

比如说，如果提供了六个 TopicPartitionOffset 实例且 concurrency 为 3；则每个容器分到两个分区。对于五个 TopicPartitionOffset 实例，两个容器各分到两个分区，第三个分到一个。如果 concurrency 大于 TopicPartition 的数量，则会下调 concurrency，使每个容器分到一个分区。

备注

client.id 属性（如果设置）会加上后缀 -n，其中 n 是与并发对应的 consumer 实例。当启用 JMX 时，这是为 MBean 提供唯一名称所必需的。

从版本 1.3 开始，MessageListenerContainer 提供了对底层 KafkaConsumer 指标的访问。对于 ConcurrentMessageListenerContainer，metrics() 方法返回所有目标 KafkaMessageListenerContainer 实例的指标。这些指标按为底层 KafkaConsumer 提供的 client-id 分组到 Map<MetricName, ? extends Metric 中。

从版本 2.3 开始，ContainerProperties 提供了一个 idleBetweenPolls 选项，让监听器容器中的主循环在 KafkaConsumer.poll() 调用之间休眠。实际的休眠间隔取该选项与 max.poll.interval.ms consumer 配置和当前记录批处理时间之差的较小值。

提交偏移量

提供了几种提交偏移量的选项。如果 enable.auto.commit consumer 属性为 true，Kafka 会根据其配置自动提交偏移量。如果为 false，容器支持多种 AckMode 设置（在下一个列表中描述）。默认的 AckMode 是 BATCH。从版本 2.3 开始，除非在配置中显式设置，框架会将 enable.auto.commit 设置为 false。以前，如果未设置该属性，则使用 Kafka 默认值（true）。

consumer 的 poll() 方法返回一个或多个 ConsumerRecords。对每条记录都会调用 MessageListener。以下列表描述了容器对每种 AckMode 所采取的动作（当不使用事务时）：

• RECORD：当监听器处理完记录返回时提交该记录的偏移量。
• BATCH：当 poll() 返回的所有记录都已处理完时提交偏移量。
• TIME：当 poll() 返回的所有记录都已处理完、且自上次提交以来已超过 ackTime 时提交偏移量。
• COUNT：当 poll() 返回的所有记录都已处理完、且自上次提交以来已收到 ackCount 条记录时提交偏移量。
• COUNT_TIME：类似于 TIME 和 COUNT，但只要任一条件为 true 就执行提交。
• MANUAL：由消息监听器负责 acknowledge() Acknowledgment。此后应用与 BATCH 相同的语义。
• MANUAL_IMMEDIATE：当监听器调用 Acknowledgment.acknowledge() 方法时立即提交偏移量。

使用事务时，偏移量被发送到事务中，其语义等效于 RECORD 或 BATCH，具体取决于监听器类型（记录或批）。

备注

MANUAL 和 MANUAL_IMMEDIATE 要求监听器是一个 AcknowledgingMessageListener 或 BatchAcknowledgingMessageListener。请参阅消息监听器。

根据 syncCommits 容器属性的不同，会使用 consumer 上的 commitSync() 或 commitAsync() 方法。syncCommits 默认为 true；另请参阅 setSyncCommitTimeout。请参阅 setCommitCallback 以获取异步提交的结果；默认回调是 LoggingCommitCallback，它记录错误（以及调试级别的成功）。

由于监听器容器有自己提交偏移量的机制，它更倾向于将 Kafka 的 ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG 设置为 false。从版本 2.3 开始，除非在 consumer 工厂或容器的 consumer 属性覆盖中显式设置，它会无条件地将其设置为 false。

Acknowledgment 有以下方法：

public interface Acknowledgment {

    void acknowledge();

}

此方法让监听器控制何时提交偏移量。

从版本 2.3 开始，Acknowledgment 接口有两个附加方法 nack(long sleep) 和 nack(int index, long sleep)。第一个与记录监听器一起使用，第二个与批监听器一起使用。为你的监听器类型调用错误的方法会抛出 IllegalStateException。

备注

如果你想提交部分批次，使用 nack()。使用事务时，将 AckMode 设置为 MANUAL；调用 nack() 会将成功处理的记录的偏移量发送到事务中。

important

nack() 只能在调用你监听器的 consumer 线程上调用。

important

使用乱序提交时不允许使用 nack()。

对于记录监听器，当调用 nack() 时，任何待提交的偏移量都会被提交，上一次 poll 的其余记录被丢弃，并在它们的分区上执行查找，以便失败的记录和未处理的记录在下一次 poll() 时被重新投递。通过设置 sleep 参数，可以在重新投递之前暂停 consumer。这类似于容器配置为 DefaultErrorHandler 时抛出异常的功能。

important

nack() 会在指定的休眠时长内暂停整个监听器，包括所有已分配的分区。

使用批监听器时，你可以指定批次中发生失败的索引。当调用 nack() 时，会为该索引之前的记录提交偏移量，并对失败和被丢弃记录所在的分区执行查找，以便它们在下一次 poll() 时被重新投递。

有关更多信息，请参阅容器错误处理器。

important

在休眠期间 consumer 会被暂停，以便我们继续 poll broker 以保持 consumer 存活。实际的休眠时间及其分辨率取决于容器的 pollTimeout，默认为 5 秒。最小休眠时间等于 pollTimeout，并且所有休眠时间都是它的倍数。对于较短的休眠时间，或为提高其精度，请考虑减小容器的 pollTimeout。

从版本 3.0.10 开始，批监听器可以使用 Acknowledgment 参数上的 acknowledge(index) 来提交批次中部分记录的偏移量。调用此方法时，该索引处记录的偏移量（以及之前所有记录的偏移量）将被提交。在执行部分批次提交之后调用 acknowledge() 将提交批次其余部分的偏移量。适用以下限制：

• 需要 AckMode.MANUAL_IMMEDIATE
• 该方法必须在监听器线程上调用
• 监听器必须消费 List 而不是原始的 ConsumerRecords
• 索引必须在列表元素的范围内
• 索引必须大于先前调用中使用的索引

这些限制会被强制执行，违反时该方法将抛出 IllegalArgumentException 或 IllegalStateException，具体取决于违反情况。

监听器容器自动启动

监听器容器实现了 SmartLifecycle，autoStartup 默认为 true。这些容器在一个较晚的阶段启动（Integer.MAX-VALUE - 100）。其他实现 SmartLifecycle、用于处理来自监听器数据的组件，应在较早的阶段启动。- 100 为后续阶段留出了空间，使组件能够在容器之后自动启动。