自定义Spring Kafka配置注解启动报错，Bean无法注入求助

我想通过自定义注解封装Kafka配置，打造公共库来简化所有应用的Kafka配置、消除重复代码。计划在主应用类添加@CustomEnableKafka注解，自动完成Kafka监听器和生产者的配置，但自定义配置类初始化晚于Spring组件，导致@Autowired注入指定名称的KafkaTemplate时出现“找不到对应Bean”的错误。

相关代码 #

自定义注解 #

@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Import(KafkaListenerConfigurationSelector.class)
public @interface CustomEnableKafka {}

配置选择器 #

public class KafkaListenerConfigurationSelector implements DeferredImportSelector {

    @Override
    public String[] selectImports(AnnotationMetadata importingClassMetadata) {
        return new String[]{CustomKafkaAutoConfiguration.class.getName()};
    }
}

自定义配置类 #

@Slf4j
@Configuration
@EnableConfigurationProperties(CustomKafkaPropertiesMap.class)
@AutoConfigureBefore({KafkaAutoConfiguration.class})
@RequiredArgsConstructor
public class CustomKafkaAutoConfiguration {

    // 来自application.yml的配置属性
    private final CustomKafkaPropertiesMap propertiesMap;
    private final ConfigurableListableBeanFactory configurableListableBeanFactory;

    @PostConstruct
    public void postProcessBeanFactory() {
        // 注册Bean的逻辑
        propertiesMap.forEach((configName, properties) -> {
          // 配置生产者工厂，Bean名称：myTopicKafkaProducerFactory
          var producerFactory = new DefaultKafkaProducerFactory<>(senderProps(properties));
          configurableListableBeanFactory.registerSingleton(configName + "KafkaProducerFactory", producerFactory);
    
          // 配置KafkaTemplate，Bean名称：myTopicKafkaTemplate
          var kafkaTemplate = new KafkaTemplate<>(producerFactory);
          configurableListableBeanFactory.registerSingleton(configName + "KafkaTemplate", kafkaTemplate);
       });
    }
}

注入示例 #

@Service
public class TestService {
    @Autowired
    @Qualifier("myTopicKafkaTemplate")
    private KafkaTemplate<String, Object> myTopicKafkaTemplate;
}

报错信息 #

---

APPLICATION FAILED TO START

---

Description:

Field myTopicKafkaTemplate in com.example.demo.service.TestService required a bean of type 'org.springframework.kafka.core.KafkaTemplate' that could not be found.

The injection point has the following annotations:
- @org.springframework.beans.factory.annotation.Autowired(required=true)

解决方案 #

问题核心是@PostConstruct方法的执行时机太晚：它是在CustomKafkaAutoConfiguration实例化后才执行，此时Spring已经开始扫描并注入TestService等业务组件，导致这些组件找不到后续注册的KafkaTemplate Bean。

需要改用BeanDefinitionRegistryPostProcessor接口来提前注册Bean定义，这个接口的方法会在Spring加载Bean定义的阶段执行，早于任何Bean的实例化和依赖注入。

修改后的CustomKafkaAutoConfiguration #

@Slf4j
@Configuration
@EnableConfigurationProperties(CustomKafkaPropertiesMap.class)
@AutoConfigureBefore({KafkaAutoConfiguration.class})
@RequiredArgsConstructor
public class CustomKafkaAutoConfiguration implements BeanDefinitionRegistryPostProcessor {

    private final CustomKafkaPropertiesMap propertiesMap;

    @Override
    public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry) throws BeansException {
        propertiesMap.forEach((configName, properties) -> {
            // 注册ProducerFactory的Bean定义
            BeanDefinition producerFactoryBeanDef = BeanDefinitionBuilder.genericBeanDefinition(DefaultKafkaProducerFactory.class)
                    .addConstructorArgValue(senderProps(properties))
                    .getBeanDefinition();
            registry.registerBeanDefinition(configName + "KafkaProducerFactory", producerFactoryBeanDef);

            // 注册KafkaTemplate的Bean定义
            BeanDefinition kafkaTemplateBeanDef = BeanDefinitionBuilder.genericBeanDefinition(KafkaTemplate.class)
                    .addConstructorArgReference(configName + "KafkaProducerFactory")
                    .getBeanDefinition();
            registry.registerBeanDefinition(configName + "KafkaTemplate", kafkaTemplateBeanDef);
        });
    }

    @Override
    public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
        // 可留空，或添加其他Bean工厂处理逻辑
    }

    // 原有的senderProps方法保持不变
    private Map<String, Object> senderProps(CustomKafkaProperties properties) {
        // 你的配置转换逻辑
    }
}

关键说明 #

1. 实现BeanDefinitionRegistryPostProcessor：该接口的postProcessBeanDefinitionRegistry方法会在Spring解析完所有配置类的Bean定义后、Bean实例化之前执行，此时注册的Bean定义能被Spring扫描到，确保业务组件注入时能找到对应的Bean。
2. 使用BeanDefinitionBuilder：通过Bean定义的方式注册，而非直接创建实例，让Spring管理Bean的生命周期（比如初始化、销毁回调），比直接注册单例更符合Spring规范。
3. 保留@AutoConfigureBefore：确保自定义配置在Spring原生的KafkaAutoConfiguration之前执行，避免Bean定义冲突。

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