springboot如何開啟和關閉kafka消費

更新時間：2024年12月16日 09:44:17 作者：阿拉的夢想

在Kafka消費者中,通過關閉自動消費配置,使用自定義容器工廠,并在消費監(jiān)聽器上設置id,可以手動控制消費的開啟和關閉,這是根據(jù)個人經(jīng)驗總結(jié)的方法,旨在幫助其他開發(fā)者

springboot開啟和關閉kafka消費

關閉kafka自動消費

配置自定義容器工廠

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.stereotype.Component;

@Component
@Configuration
public class kafkaConfig {

    @Autowired
    private ConsumerFactory<String, String> consumerFactory;

    @Bean("pingKafkaFactory")
    public ConcurrentKafkaListenerContainerFactory<String, String> delayContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> container = new ConcurrentKafkaListenerContainerFactory<String, String>();
        container.setConsumerFactory(consumerFactory);
        //禁止自動啟動
        container.setAutoStartup(false);
        return container;
    }
}

在消費監(jiān)聽器上使用工廠，并設置id

@KafkaListener(topics = "#{pingProperties.getTopic().split(',')}",id = "pingConsumer",containerFactory = "pingKafkaFactory")

這樣，啟動項目后，就不會自動消費了。

手動開啟和關閉消費

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.stereotype.Service;

/**
 * Kafka消費監(jiān)聽服務實現(xiàn)類.
 */
@Service
@Slf4j
public class KafkaConsumerListenerServiceImpl implements KafkaConsumerListenerService {


    /**
     * registry.
     */
    @Autowired
    private KafkaListenerEndpointRegistry registry;

    /**
     * 開啟監(jiān)聽.
     *
     * @param listenerId 監(jiān)聽ID
     */
    @Override
    public void startListener(String listenerId) {
        //判斷監(jiān)聽容器是否啟動，未啟動則將其啟動
        if (!registry.getListenerContainer(listenerId).isRunning()) {
            registry.getListenerContainer(listenerId).start();
        }
        //項目啟動的時候監(jiān)聽容器是未啟動狀態(tài)，而resume是恢復的意思不是啟動的意思
        //registry.getListenerContainer(listenerId).stop();
        log.info(listenerId + "開啟監(jiān)聽成功。");
    }

    /**
     * 停止監(jiān)聽.
     *
     * @param listenerId 監(jiān)聽ID
     */
    @Override
    public void stopListener(String listenerId) {
        registry.getListenerContainer(listenerId).stop();
        log.info(listenerId + "停止監(jiān)聽成功。");
    }

}