如何通过全局标志禁用Kafka Consumer

问题描述 #

需要通过环境变量KAFKA_CONSUMER_ENABLED控制Kafka Consumer的启停，已在配置类中添加条件：仅当标志为true时才创建容器工厂，但Kafka Consumer仍尝试创建实例，需解决该问题。

提供的代码如下：

配置类代码 #

@Configuration
@EnableKafka
public class KafkaConsumerConfig {

  public static final String LOCALHOST_9200 = "localhost:9200";
  public static final String KAFKA_BROKERS = "KAFKA_BROKERS";
  private final String KAFKA_ENDPOINT =
      StringUtils.isNullOrEmpty(System.getenv(KAFKA_BROKERS)) ? LOCALHOST_9200
          : System.getenv(KAFKA_BROKERS);

  private final boolean KAFKA_CONSUMER_ENABLED = Boolean.parseBoolean(
      System.getenv("KAFKA_CONSUMER_ENABLED"));

  @Bean
  public ConsumerFactory<String, String> consumerFactory() {
    System.out.println(System.getenv(KAFKA_BROKERS));
    System.out.println(KAFKA_CONSUMER_ENABLED);
    System.out.println("KafkEndpoint :: "+ KAFKA_ENDPOINT);
    if (!KAFKA_CONSUMER_ENABLED) {
      return null;
    }
    Map<String, Object> config = new HashMap<>();
    config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_ENDPOINT);
    config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    return new DefaultKafkaConsumerFactory<>(config);
  }

  @Bean
  public ConcurrentKafkaListenerContainerFactory<String, String> concurrentKafkaListenerContainerFactory() {
    if (!KAFKA_CONSUMER_ENABLED) {
      return null;
    }
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory();
    factory.setConsumerFactory(consumerFactory());
    return factory;
  }
}

消费者类代码 #

@Service
@Slf4j
public class KafkaPayrollDocUpdateConsumer {

  private final boolean KAFKA_CONSUMER_ENABLED = Boolean.parseBoolean(
      System.getenv("KAFKA_CONSUMER_ENABLED"));

  @KafkaListener(topics = "#{T(java.lang.System).getenv('NODE_ENV') + '-doc-centre-doc-create-update-event'}", groupId = "doc-centre-doc-create-update-event", containerFactory = "concurrentKafkaListenerContainerFactory")
  public void listen(String message) {
    

  }
}

问题原因 #

当前写法存在两个核心问题：

1. 直接返回null的Bean不符合Spring Bean管理规范，Spring依然会尝试处理该Bean的依赖关系，可能引发异常
2. @KafkaListener注解会被Spring扫描到，即使容器工厂不存在，依然会尝试创建消费者容器，导致无效实例或报错

解决方案 #

方案一：用@Conditional控制Bean创建 #

通过自定义条件注解，只有当KAFKA_CONSUMER_ENABLED为true时才创建Consumer相关Bean，彻底避免无效Bean的实例化。

修改配置类：

@Configuration
@EnableKafka
public class KafkaConsumerConfig {

  public static final String LOCALHOST_9200 = "localhost:9200";
  public static final String KAFKA_BROKERS = "KAFKA_BROKERS";
  private final String KAFKA_ENDPOINT =
      StringUtils.isNullOrEmpty(System.getenv(KAFKA_BROKERS)) ? LOCALHOST_9200
          : System.getenv(KAFKA_BROKERS);

  // 自定义条件判断类
  static class KafkaConsumerEnabledCondition implements Condition {
    @Override
    public boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {
      return Boolean.parseBoolean(context.getEnvironment().getProperty("KAFKA_CONSUMER_ENABLED"));
    }
  }

  @Bean
  @Conditional(KafkaConsumerEnabledCondition.class)
  public ConsumerFactory<String, String> consumerFactory() {
    System.out.println(System.getenv(KAFKA_BROKERS));
    System.out.println("KafkEndpoint :: "+ KAFKA_ENDPOINT);
    Map<String, Object> config = new HashMap<>();
    config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_ENDPOINT);
    config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    return new DefaultKafkaConsumerFactory<>(config);
  }

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

方案二：控制@KafkaListener自动启动 #

在@KafkaListener中通过autoStartup属性，根据环境变量控制监听器是否自动启动，即使Bean存在也不会触发消费逻辑。

修改消费者类：

@Service
@Slf4j
public class KafkaPayrollDocUpdateConsumer {

  @KafkaListener(
      topics = "#{T(java.lang.System).getenv('NODE_ENV') + '-doc-centre-doc-create-update-event'}",
      groupId = "doc-centre-doc-create-update-event",
      containerFactory = "concurrentKafkaListenerContainerFactory",
      autoStartup = "#{@environment.getProperty('KAFKA_CONSUMER_ENABLED', 'false') == 'true'}"
  )
  public void listen(String message) {
    // 业务逻辑
  }
}

方案三：结合@Profile按环境控制 #

如果是基于不同环境（如开发/生产）控制，可以用@Profile指定只有激活特定环境时才加载相关配置和消费者。

修改配置类：

@Configuration
@EnableKafka
@Profile("kafka-enabled")
public class KafkaConsumerConfig {
  // 原有代码不变
}

修改消费者类：

@Service
@Slf4j
@Profile("kafka-enabled")
public class KafkaPayrollDocUpdateConsumer {
  // 原有代码不变
}

最佳实践 #

推荐方案一+方案二结合使用：

1. 用@Conditional避免禁用时创建不必要的Bean，减少资源消耗
2. 用autoStartup确保即使Bean意外创建，监听器也不会启动消费逻辑

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