跳到主要内容
版本:5.0.1

消费批次

DeepSeek V3 中英对照 Consuming Batches

从版本 3.0 开始,当 spring.cloud.stream.bindings.<name>.consumer.batch-mode 设置为 true 时,通过轮询 Kafka Consumer 接收的所有记录都将以 List<?> 的形式呈现给监听器方法。否则,该方法将每次处理一条记录。批处理的大小由 Kafka 消费者属性 max.poll.records、fetch.min.bytes 和 fetch.max.wait.ms 控制;更多信息请参阅 Kafka 文档。

接收批次时,允许以下类型签名:

List<Person>
Message<List<Person>>

在 List<Person> 的第一种选项中,监听器将不会收到任何消息头。如果使用第二种类型签名(Message<List<Person>>),那么就可以访问消息头;然而,所有消息头仍然以 Collection 的形式存在。让我们来看下面的例子。

假设 Message 包含一个包含十个 Person 对象的列表。Message 的 MessageHeaders 包含一个头部映射,其中键为头部名称,值为列表。该列表按与有效载荷列表相同的顺序包含该头部的值。因此,应用程序需要根据有效载荷列表的迭代,从 MessageHeaders 映射中正确访问头部。

请注意,在批处理模式下使用时,不允许使用 List<Message<Person>> 这种形式的类型签名。

从版本 4.0.2 开始,binder 在批处理模式下消费时支持 DLQ 功能。请注意,当在批处理模式的消费者绑定上使用 DLQ 时,从前一次轮询接收到的所有记录都将被发送到 DLQ 主题。

important

在批处理模式下不支持在绑定器内部进行重试,因此 maxAttempts 将被覆盖为 1。你可以配置一个 DefaultErrorHandler(使用 ListenerContainerCustomizer)来实现类似于绑定器中重试的功能。你也可以使用手动 AckMode 并调用 Ackowledgment.nack(index, sleep) 来提交部分批次的偏移量,并使剩余记录重新投递。有关这些技术的更多信息,请参阅 Spring for Apache Kafka 文档。

备注

在批处理模式下接收 KafkaNull 对象时,接收到的列表将包含一个对应 KafkaNull 对象的 null 元素。这对于 List<Person> 和 Message<List<Person>> 两种类型的签名都适用。

批量模式消费时的可观测性​

在批量消费记录时,观测追踪传播功能无法直接支持。这是因为 Kafka 绑定器使用的 Spring for Apache Kafka 库不支持批量监听器的追踪功能,仅支持记录监听器。在批量监听器中,接收到的记录可能来自多个主题/分区以及多个生产者,而这些生产者添加追踪信息是可选的。由于批次中的记录之间可能没有任何关联,框架无法对它们进行追踪做出任何假设,例如为它们提供单一的追踪 ID 等。如果你使用 Message<List<String>> 类型签名,你可以获取一个名为 kafka_batchConvertedHeaders 的头部信息,其中包含一个与你的负载条目数量相同的列表。这个列表包含一个 Map,其中存储了追踪头部信息。然而,需要应用程序自行正确遍历这些信息并启动观测。