如何在自定义DeadLetterPublishingRecoverer中注入Spring @Value配置？

你的核心矛盾是：自定义DeadLetterPublishingRecoverer既需要读取Spring配置，又不能直接加组件注解（否则构造函数的BiFunction参数会因无对应Bean报错）。以下是几种可行方案：

方案1：用@Configuration类手动注册Bean（推荐） #

通过配置类获取配置值，同时手动传入构造函数所需的BiFunction，让自定义Recoverer成为Spring管理的Bean，同时避免自动注入冲突。

步骤1：编写自定义Recoverer类（无需加组件注解） #

public class CustomDLTRecoverer extends DeadLetterPublishingRecoverer {
    private final String myValue;

    // 构造函数接收Kafka操作类、目标解析函数、配置值
    public CustomDLTRecoverer(KafkaOperations<?, ?> kafkaOperations,
                              BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> destinationResolver,
                              String myValue) {
        super(kafkaOperations, destinationResolver);
        this.myValue = myValue;
    }

    @Override
    protected ProducerRecord<?, ?> createProducerRecord(ConsumerRecord<?, ?> consumerRecord,
                                                       Exception exception,
                                                       TopicPartition topicPartition) {
        // 在这里使用myValue生成符合DLT Schema的消息
        ProducerRecord<?, ?> originalRecord = super.createProducerRecord(consumerRecord, exception, topicPartition);
        // 示例：添加自定义header或修改payload
        return new ProducerRecord<>(
                topicPartition.topic(),
                topicPartition.partition(),
                originalRecord.timestamp(),
                originalRecord.key(),
                modifyPayloadWithConfig(originalRecord.value(), myValue)
        );
    }

    private Object modifyPayloadWithConfig(Object originalPayload, String configValue) {
        // 自定义消息处理逻辑，比如拼接配置值到payload
        return originalPayload + "_dlt_" + configValue;
    }
}

步骤2：编写配置类，创建Recoverer Bean #

@Configuration
public class KafkaRecoveryConfig {

    // 从application.yml注入配置值
    @Value("${your.config.key}")
    private String myConfigValue;

    @Bean
    public CustomDLTRecoverer customDLTRecoverer(KafkaOperations<?, ?> kafkaOperations) {
        // 手动定义构造函数所需的BiFunction（根据你的业务逻辑实现）
        BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> destinationResolver = (record, ex) -> {
            // 示例：将死信发送到对应主题的DLT分区
            return new TopicPartition(record.topic() + ".dlt", record.partition());
        };

        // 手动实例化自定义Recoverer，传入所有构造参数
        return new CustomDLTRecoverer(kafkaOperations, destinationResolver, myConfigValue);
    }
}

方案2：实现ApplicationContextAware获取配置 #

如果不想通过构造函数传递配置，可以让自定义Recoverer实现ApplicationContextAware接口，直接从Spring环境中读取配置：

public class CustomDLTRecoverer extends DeadLetterPublishingRecoverer implements ApplicationContextAware {
    private String myValue;

    public CustomDLTRecoverer(KafkaOperations<?, ?> kafkaOperations,
                              BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> destinationResolver) {
        super(kafkaOperations, destinationResolver);
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        // 从Spring环境中读取配置
        this.myValue = applicationContext.getEnvironment().getProperty("your.config.key");
    }

    @Override
    protected ProducerRecord<?, ?> createProducerRecord(ConsumerRecord<?, ?> consumerRecord,
                                                       Exception exception,
                                                       TopicPartition topicPartition) {
        // 使用myValue处理消息，逻辑同方案1
        // ...
    }
}

然后同样在KafkaRecoveryConfig中手动创建Bean：

@Configuration
public class KafkaRecoveryConfig {
    @Bean
    public CustomDLTRecoverer customDLTRecoverer(KafkaOperations<?, ?> kafkaOperations) {
        BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> resolver = (record, ex) -> {
            // 自定义目标解析逻辑
            return new TopicPartition("dlt-topic", 0);
        };
        return new CustomDLTRecoverer(kafkaOperations, resolver);
    }
}

方案3：用@ConfigurationProperties绑定配置类（多配置场景推荐） #

如果需要读取多个配置项，建议用@ConfigurationProperties封装配置，更易维护：

步骤1：定义配置类 #

@ConfigurationProperties(prefix = "dlt")
public class DLTConfig {
    private String myValue;
    // 可添加其他配置字段，比如dltTopicSuffix、retryCount等

    // getter和setter
    public String getMyValue() {
        return myValue;
    }

    public void setMyValue(String myValue) {
        this.myValue = myValue;
    }
}

步骤2：配置类中启用并注入 #

@Configuration
@EnableConfigurationProperties(DLTConfig.class)
public class KafkaRecoveryConfig {

    @Bean
    public CustomDLTRecoverer customDLTRecoverer(KafkaOperations<?, ?> kafkaOperations,
                                                 DLTConfig dltConfig) {
        BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> resolver = (record, ex) -> {
            return new TopicPartition(record.topic() + ".dlt", record.partition());
        };
        return new CustomDLTRecoverer(kafkaOperations, resolver, dltConfig.getMyValue());
    }
}

核心原理 #

以上方案的本质是：通过@Configuration类手动控制CustomDLTRecoverer的Bean创建过程，既让它纳入Spring上下文以获取配置，又避免了Spring自动注入构造函数参数的冲突（因为BiFunction是我们手动传入的，不需要Spring提供对应的Bean）。

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