快捷導(dǎo)航

springboot如何開啟和關(guān)閉kafka消費(fèi)

更新時(shí)間：2024年12月16日 09:44:17 作者：阿拉的夢(mèng)想

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

springboot開啟和關(guān)閉kafka消費(fèi)

關(guān)閉kafka自動(dòng)消費(fèi)

配置自定義容器工廠

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);
        //禁止自動(dòng)啟動(dòng)
        container.setAutoStartup(false);
        return container;
    }
}

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

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

這樣，啟動(dòng)項(xiàng)目后，就不會(huì)自動(dòng)消費(fèi)了。

手動(dòng)開啟和關(guān)閉消費(fèi)

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

/**
 * Kafka消費(fèi)監(jiān)聽服務(wù)實(shí)現(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)聽容器是否啟動(dòng)，未啟動(dòng)則將其啟動(dòng)
        if (!registry.getListenerContainer(listenerId).isRunning()) {
            registry.getListenerContainer(listenerId).start();
        }
        //項(xiàng)目啟動(dòng)的時(shí)候監(jiān)聽容器是未啟動(dòng)狀態(tài)，而resume是恢復(fù)的意思不是啟動(dòng)的意思
        //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)聽成功。");
    }

}