如何为@KafkaListener配置自定义方法参数传递？

Spring Kafka的@KafkaListener方法默认只支持特定类型的参数（如ConsumerRecord、Acknowledgment等），无法直接将自定义的CustomContext作为方法参数传入，但可以通过以下几种方式实现你的需求：

方案1：依赖注入（适合全局/单例上下文） #

如果你的CustomContext是全局共享的单例对象，直接通过Spring的依赖注入将其注入到消费者类中即可，无需作为方法参数传递。

代码示例：

@Component
public class TestKafkaConsumer {
    private final CustomContext customContext;
    private final Logger log = LoggerFactory.getLogger(TestKafkaConsumer.class);

    // 推荐使用构造注入
    public TestKafkaConsumer(CustomContext customContext) {
        this.customContext = customContext;
    }

    @KafkaListener(
            id = "TestService",
            topics = "testTpoc",
            groupId = "kafkaGroupId1",
            autoStartup = "true")
    public void receiveTestEventWithContext(final ConsumerRecord<String, String> consumerRecord) {
        log.info("Received kafka message with custom context");
        // 直接使用customContext调用下游API
        downstreamApi.call(customContext);
    }
}

方案2：ThreadLocal + ConsumerInterceptor（适合请求级上下文） #

如果每个消费请求需要独立的CustomContext（比如根据消息内容生成上下文），可以通过ConsumerInterceptor在消费前将上下文存入ThreadLocal，然后在方法中取出使用。

1. 定义上下文持有类：

public class CustomContextHolder {
    private static final ThreadLocal<CustomContext> CONTEXT_HOLDER = new ThreadLocal<>();

    public static void setCustomContext(CustomContext context) {
        CONTEXT_HOLDER.set(context);
    }

    public static CustomContext getCustomContext() {
        return CONTEXT_HOLDER.get();
    }

    public static void clear() {
        CONTEXT_HOLDER.remove();
    }
}

2. 实现ConsumerInterceptor：

public class CustomContextInterceptor implements ConsumerInterceptor<String, String> {

    @Override
    public ConsumerRecord<String, String> onConsume(ConsumerRecord<String, String> record) {
        // 这里可以根据record的内容、消息头或其他逻辑生成CustomContext
        CustomContext context = new CustomContext();
        context.setTraceId(new String(record.headers().lastHeader("traceId").value()));
        CustomContextHolder.setCustomContext(context);
        return record;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
        // 消费完成后清理ThreadLocal，避免内存泄漏
        CustomContextHolder.clear();
    }

    @Override
    public void close() {}

    @Override
    public void configure(Map<String, ?> configs) {}
}

3. 在Kafka消费者配置中添加拦截器：

@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-server:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "kafkaGroupId1");
        // 注册自定义拦截器
        props.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, CustomContextInterceptor.class.getName());
        // 其他必要配置...
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

4. 在消费方法中使用上下文：

@Component
public class TestKafkaConsumer {
    private final Logger log = LoggerFactory.getLogger(TestKafkaConsumer.class);

    @KafkaListener(
            id = "TestService",
            topics = "testTpoc",
            groupId = "kafkaGroupId1",
            autoStartup = "true")
    public void receiveTestEventWithContext(final ConsumerRecord<String, String> consumerRecord) {
        CustomContext customContext = CustomContextHolder.getCustomContext();
        log.info("Received kafka message with custom context");
        // 用customContext调用下游API
        downstreamApi.call(customContext);
    }
}

方案3：消息头 + @Header注解（适合与消息绑定的上下文） #

如果CustomContext是和具体消息绑定的（比如发送消息时附带的上下文），可以将其序列化后放入消息头，消费时通过@Header注解取出并反序列化。

1. 发送消息时添加上下文到消息头：

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
@Autowired
private ObjectMapper objectMapper;

public void sendMessage(String message, CustomContext customContext) throws JsonProcessingException {
    String contextJson = objectMapper.writeValueAsString(customContext);
    kafkaTemplate.send("testTpoc", message)
            .addHeader("customContext", contextJson.getBytes(StandardCharsets.UTF_8));
}

2. 消费方法中通过@Header获取上下文：

@Component
public class TestKafkaConsumer {
    private final Logger log = LoggerFactory.getLogger(TestKafkaConsumer.class);
    @Autowired
    private ObjectMapper objectMapper;

    @KafkaListener(
            id = "TestService",
            topics = "testTpoc",
            groupId = "kafkaGroupId1",
            autoStartup = "true")
    public void receiveTestEventWithContext(
            @Header(name = "customContext", required = false) byte[] contextBytes,
            final ConsumerRecord<String, String> consumerRecord) throws IOException {
        CustomContext customContext = objectMapper.readValue(contextBytes, CustomContext.class);
        log.info("Received kafka message with custom context");
        // 下游API调用
        downstreamApi.call(customContext);
    }
}

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