Spring Cloud Stream Kafka批量消费如何获取含Header的Message列表

核心原因 #

你当前配置缺少Kafka binder层级的批量消息头映射配置，Spring Cloud Stream Kafka默认批量消费时如果没有显式开启头传递，会自动提取payload生成列表，忽略外层的Message包装。 #

修复步骤 #

1. 修正配置缩进并新增Kafka专属消费者配置 #

首先修正你贴出的yaml缩进错误（原配置中cloud和spring同级，语法错误），同时新增Kafka binder专属的批量配置和头映射配置：

spring:
  cloud:
    function:
      definition: function
    stream:
      default-binder: my-avro-binder
      bindings:
        function-in-0:
          binder: my-avro-binder
          destination: function-output
          group: constant-name
          contentType: application/*+avro
          consumer:
            useNativeEncoding: true
            batchMode: true
            headerMode: headers
      # 新增kafka binder专属配置
      kafka:
        bindings:
          function-in-0:
            consumer:
              batch-mode: true
              header-mapper: customKafkaHeaderMapper

2. 注册HeaderMapper Bean放行自定义头 #

默认HeaderMapper只会传递Kafka标准头，自定义头需要显式配置放行：

import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class KafkaHeaderConfig {
    @Bean
    public DefaultKafkaHeaderMapper customKafkaHeaderMapper() {
        DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
        // 按需添加允许传递的头名称，支持通配符，生产环境不建议直接用*
        mapper.setAllowedHeaders("*");
        return mapper;
    }
}

3. 调整Function输入类型 #

直接显式声明输入为List<Message<MyType>>，不要用通配符List<?>避免类型推断错误：

import org.springframework.messaging.Message;
import java.util.List;
import java.util.function.Function;

@Bean
public Function<List<Message<MyType>>, List<Message<MyType>>> function() {
    return list -> {
        // 遍历即可拿到每个消息的完整头和payload
        list.forEach(item -> {
            // item.getHeaders() 拿到所有头信息
            // item.getPayload() 拿到业务数据
        });
        // 业务逻辑
        return list;
    };
}

验证注意项 #

• Spring Cloud Stream版本需 >= 3.1，旧版本存在批量Message传递的已知bug
• 不要配置全局的spring.cloud.stream.kafka.binder.batch-mode，优先使用binding级别的配置避免冲突
• 确认生产者侧已将自定义头正确写入Kafka记录的头字段，而非写入payload内部

内容的提问来源于stack exchange，提问作者Yosi Bronsberg