spring kafka 中启用批处理模式后，@kafkalistener 方法需明确声明接收 list 类型参数，否则仅解析首条消息；本文详解配置要点、方法签名修正及常见陷阱。
在 Spring Kafka 中实现批量消费（batch consumption）时，一个常见误区是：虽然已正确配置 setBatchListener(true) 并设置了 max.poll.records=5，但 @KafkaListener 方法仍只接收到单条消息——这并非 Kafka 或序列化层的问题，而是 Spring Kafka 的编程模型约束所致。
✅ 正确的批处理监听器签名
当启用批处理模式（factory.setBatchListener(true)）后，Spring Kafka 不会将每条反序列化后的记录单独调用一次监听方法，而是将整个 ConsumerRecords
@KafkaListener(
topics = "#{'${my.kafka.conf.topics}'.split(',')}",
concurrency = "${my.kafka.conf.concurrency}",
clientIdPrefix = "${my.kafka.conf.clientIdPrefix}",
groupId = "${my.kafka.conf.groupId}"
)
public void kafkaListener(
List<Flight> flights, // ✅ 关键：改为 List<Flight>
@Header(KafkaHeaders.OFFSET) List<Long> offsets,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
@Header(KafkaHeaders.RECEIVED_TIMESTAMP) List<Long> timestamps) {
if (flights == null || flights.isEmpty()) return;
// 按索引一一对应（offsets.get(i) 对应 flights.get(i)）
for (int i = 0; i < flights.size(); i++) {
Flight flight = flights.get(i);
Long offset = offsets.get(i);
Integer partition = partitions.get(i);
Long timestamp = timestamps.get(i);
logger.info("Processing flight {} at offset {} on partition {}",
flight.getFlightId(), offset, partition);
// TODO: 业务逻辑处理
}
}⚠️ 注意：所有 @Header 参数也必须声明为 List 类型，且与 List 元素严格一一对应（顺序一致），这是 Spring Kafka 批处理的契约要求。
? 配置验证要点
你当前的配置基本正确，但仍需确认以下关键项：
- ✅ ConcurrentKafkaListenerContainerFactory#setBatchListener(true) 已调用（已在代码中体现）；
- ✅ spring.kafka.listener.type 在 YAML 中设为 single 是合理的（表示单消费者实例内支持批处理），但更推荐显式统一使用 batch 类型（Spring Boot 3.2+ 支持）；
- ❗ spring.kafka.listener.ack-mode: batch 必须与 setBatchListener(true) 匹配，否则可能引发提交行为异常；
- ❗ 自定义 KafkaCustomDeserializer 必须线程安全，且其 deserialize(...) 方法不能缓存或复用返回对象（避免多条消息被同一实例覆盖）；建议每次新建对象或使用 ThreadLocal 隔离。
? 补充说明：为什么日志显示“5 条消息被反序列化”？
你在自定义反序列化器中加日志看到 5 次输出，是因为 Kafka Consumer 确实拉取了 5 条原始字节数据，并由 KafkaCustomDeserializer 分别调用了 5 次 deserialize() —— 这属于底层反序列化阶段，不等于监听器已接收到 5 个 Java 对象。Spring Kafka 在此之后会将这些对象组装为 List
✅ 最终检查清单
| 项目 | 是否满足 | 说明 |
|---|---|---|
| @KafkaListener 方法参数为 List |
✅ 必须 | 否则无法触发批处理分发逻辑 |
| @Header 参数均为 List |
✅ 必须 | 与消息列表索引对齐 |
| ConcurrentKafkaListenerContainerFactory.setBatchListener(true) | ✅ 已配置 | 核心开关 |
| spring.kafka.listener.ack-mode= batch | ✅ 推荐 | 确保偏移量按批次提交 |
| 自定义 Deserializer 无状态/线程安全 | ⚠️ 必须自查 | 避免对象复用导致数据污染 |
完成上述调整后，重启应用即可稳定接收完整批次消息。如仍有问题，可启用 logging.level.org.springframework.kafka=DEBUG 查看 BatchMessagingMessageListenerAdapter 的实际调用日志，进一步定位分发环节行为。
