Spring Boot에서 Kafka 소비자 활성화/비활성화 제어

Spring Boot에서 Kafka 소비자 활성화/비활성화 제어

스프링 부트에서 여러 카프카 소비자를 구성했습니다.kafka.properties는 다음과 같습니다(여기에는 한 소비자에 대한 구성만 나열됨).

kafka.topics=
bootstrap.servers=
group.id=
enable.auto.commit=
auto.commit.interval.ms=
session.timeout.ms=
schema.registry.url=
auto.offset.reset=
kafka.enabled=

다음은 구성입니다.

@Configuration
@PropertySource({"classpath:kafka.properties"})
public class KafkaConsumerConfig {

    @Autowired
    private Environment env;

    @Bean
    public ConsumerFactory<String, String> pindropConsumerFactory() {
        Map<String, Object> dataRiverProps = new HashMap<>();

        dataRiverProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, env.getProperty("bootstrap.servers"));
        dataRiverProps.put(ConsumerConfig.GROUP_ID_CONFIG, env.getProperty("group.id"));
        dataRiverProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, env.getProperty("enable.auto.commit"));
        dataRiverProps.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, env.getProperty("auto.commit.interval.ms"));
        dataRiverProps.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, env.getProperty("session.timeout.ms"));

        dataRiverProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        dataRiverProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        dataRiverProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, env.getProperty("auto.offset.reset"));

        return new DefaultKafkaConsumerFactory<>(dataRiverProps);
    }

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

그리고 이것은 소비자입니다.

@Component
public class KafkaConsumer {

    @Autowired
    private MessageProcessor messageProcessor;


    @KafkaListener(topics = "#{'${kafka.topics}'.split(',')}", containerFactory = "kafkaListenerContainerFactory")
    public void consumeJson(String message) {
        // processing message
    }
}

이 소비자의 작성 또는 메시지 검색을 제어할 수 있도록 "kafka.enabled"라는 프롭을 사용할 수 있는 방법이 있습니까?감사합니다!

아래와 같이 소비자에서 자동 시작(true/false) 속성을 사용하여 이를 수행할 수 있습니다.

@KafkaListener(id = "foo", topics = "Topic1", groupId = "group_id",
        containerFactory = "kafkaListenerContainerFactory",autoStartup = "${listen.auto.start:false}")
public void consume(String message) {
    //System.out.println("Consumed message: " + message);
}

Kafka 구성을 비활성화하려면 다음과 같이 합니다.

1. Kafka ConsumerConfig 주석 달기

   @ConditionalOnProperty(value = "kafka.enabled", matchIfMissing = true)

2. 제거한다.@Component에KafkaConsumer클래스를 지정하고 @빈 인으로 정의합니다.KafkaConsumerConfig.

KafkaConsumer에서 메시지 검색을 제어하려면 다음과 같이 하십시오.

1. Kafka Consumer 내부의 자산 가치만 확인할 수 있습니다.@Value("kafka.enabled") private Boolean enabled;

2. 그런 다음 간단한 if 방법으로 주석을 달면 사용합니다.@KafkaListener.

언급URL : https://stackoverflow.com/questions/54426658/control-enabling-disabling-kafka-consumers-in-spring-boot