@Profile注解对@KafkaListener不生效问题求助

问题根源 #

@Profile注解直接加在@KafkaListener方法上不会生效，因为Spring的@Profile在方法层面仅对**@Bean标注的工厂方法**起作用，用来控制是否创建对应的Bean。而@KafkaListener方法是由KafkaListenerAnnotationBeanPostProcessor单独扫描并注册为消息监听器的，这个处理器不会检查方法上的@Profile注解，所以无论当前激活的profile是否匹配，监听器都会被正常注册并启动消费。

可行解决方案 #

方案1：将@Profile移到类级别 #

因为你的类标注了@Service，把@Profile加到类上会控制整个Bean的创建。当profile不匹配时，整个Service Bean不会被实例化，里面的@KafkaListener方法自然也不会被注册：

@Service
@Profile("training")
public class YourConsumerService {

    @KafkaListener(id = "my_grp", topicPartitions = {
            @TopicPartition(topic = KafkaTopicConstants.MY_TOPIC, partitions = {"0", "1", "2","3","4","5"})
    }, concurrency = KafkaTopicConstants.PARTITION_COUNT)
    public void consumeRequest(ConsumerRecord<String,RequestedUrlData> record) {
        try {
            log.error("Consuming Data   :::: ");
        } catch (Exception e) {
            log.error("Exception while consuming ",e);
        }
    }
}

方案2：使用@KafkaListener的autoStartup属性结合SpEL #

如果不想控制整个Service类，只想单独控制这个监听器的启停，可以利用autoStartup属性，通过Spring表达式判断当前激活的profile是否匹配：

@Service
public class YourConsumerService {

    @KafkaListener(id = "my_grp", 
                   topicPartitions = {
                       @TopicPartition(topic = KafkaTopicConstants.MY_TOPIC, partitions = {"0", "1", "2","3","4","5"})
                   }, 
                   concurrency = KafkaTopicConstants.PARTITION_COUNT,
                   autoStartup = "#{environment.acceptsProfiles('training')}")
    public void consumeRequest(ConsumerRecord<String,RequestedUrlData> record) {
        try {
            log.error("Consuming Data   :::: ");
        } catch (Exception e) {
            log.error("Exception while consuming ",e);
        }
    }
}

这个表达式会检查当前环境是否激活了training profile，只有匹配时才会启动监听器。

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