消息监听器容器

提供了以下构造函数

它接收一个 ConsumerFactory 以及主题和分区的信息，以及其他配置，这些都在 ContainerProperties 对象中。ContainerProperties 提供了以下构造函数

第一个构造函数接受一个 TopicPartitionOffset 参数数组，以明确指示容器使用哪些分区（使用 consumer 的 assign() 方法）以及可选的初始偏移量。正值默认是绝对偏移量。负值默认是相对于分区内当前最后一个偏移量的相对值。提供了 TopicPartitionOffset 的一个构造函数，它接受一个额外的 boolean 参数。如果此参数为 true，则初始偏移量（正或负）是相对于此消费者当前位置的相对值。偏移量在容器启动时应用。第二个接受一个主题数组，Kafka 根据 group.id 属性分配分区——在组内分布分区。第三个使用正则表达式 Pattern 选择主题。

要将 MessageListener 分配给容器，您可以在创建容器时使用 ContainerProps.setMessageListener 方法。以下示例展示了如何操作

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

有关 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 开始，创建消费者工厂时，如果您以对象形式（在构造函数中或通过 setter）提供反序列化器，工厂将调用 configure() 方法，并使用配置属性对其进行配置。

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

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

如果容器属性配置为主题（或主题模式），Kafka 会使用其组管理功能在消费者之间分配分区。

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

例如，如果提供了六个 TopicPartitionOffset 实例，且 concurrency 为 3；每个容器将获得两个分区。对于五个 TopicPartitionOffset 实例，两个容器将获得两个分区，第三个获得一个。如果 concurrency 大于 TopicPartitions 的数量，则 concurrency 会向下调整，使得每个容器获得一个分区。

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

从版本 2.3 开始，ContainerProperties 提供了一个 idleBetweenPolls 选项，允许监听器容器中的主循环在 KafkaConsumer.poll() 调用之间休眠。实际的休眠间隔是提供的选项以及 max.poll.interval.ms 消费者配置与当前记录批处理时间差的最小值。

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

消费者 poll() 方法返回一个或多个 ConsumerRecords。MessageListener 会为每条记录调用一次。以下列表描述了容器在每种 AckMode 下采取的操作（不使用事务时）

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

根据 syncCommits 容器属性，使用消费者上的 commitSync() 或 commitAsync() 方法。syncCommits 默认值为 true；另请参阅 setSyncCommitTimeout。参见 setCommitCallback 以获取异步提交的结果；默认的回调是 LoggingCommitCallback，它记录错误（并在 debug 级别记录成功）。

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

Acknowledgment 提供了以下方法

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

从版本 2.3 开始，Acknowledgment 接口增加了两个额外的方法 nack(long sleep) 和 nack(int index, long sleep)。第一个用于记录监听器，第二个用于批量监听器。为您的监听器类型调用错误的方法将抛出 IllegalStateException。

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

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

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

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

这些限制将被强制执行，根据违规情况，方法将抛出 IllegalArgumentException 或 IllegalStateException。

监听器容器实现了 SmartLifecycle，并且 autoStartup 默认值为 true。容器在较晚的阶段启动（Integer.MAX-VALUE - 100）。其他实现了 SmartLifecycle 并处理来自监听器数据的组件应在更早的阶段启动。- 100 为后续阶段留出空间，以便在容器启动后自动启动组件。