springboot集成kafka消费手动启动停止

项目场景：

在月结，或者某些时候，我们需要停掉kafka所有的消费端，让其暂时停止消费，而后等月结完成，再从新对消费监听恢复，进行消费，此动作不需要重启服务，最后源码下载

解决分析

KafkaListenerEndpointRegistry这是kafka与spring集成的监听注册bean，可以通过它获取监听容器对象，然后对监听容器对象实行启动，暂停，恢复等操作

/**
 * kafka服务操作类
 * @author liangxi.zeng
 */
@Service
@Slf4j
public class KafkaService {

    @Autowired
    private KafkaListenerEndpointRegistry registry;

    /**
     * 开启消费
     * @param listenerId
     */
    public void start(String listenerId) {
        MessageListenerContainer messageListenerContainer = registry
                .getListenerContainer(listenerId);
        if(Objects.nonNull(messageListenerContainer)) {
            if(!messageListenerContainer.isRunning()) {
                messageListenerContainer.start();
            } else {
                if(messageListenerContainer.isContainerPaused()) {
                    log.info("listenerId:{},恢复",listenerId);
                    messageListenerContainer.resume();
                }
            }
        }
    }

    /**
     * 停止消费
     * @param listenerId
     */
    public void pause(String listenerId) {
        MessageListenerContainer messageListenerContainer = registry
                .getListenerContainer(listenerId);
        if(Objects.nonNull(messageListenerContainer) && !messageListenerContainer.isContainerPaused()) {
            log.info("listenerId:{},暂停",listenerId);
            messageListenerContainer.pause();
        }
    }
}

kafka启动，停止，恢复触发场景

1.通过定时任务自动触发，通过@Scheduled，在某个时间点暂停kafka某个监听的消费，也可以在某个时间点恢复kafka某个监听的消费

/**
 * 卡夫卡配置类
 * @author liangxi.zeng
 */
@Configuration
@EnableScheduling
public class KafkaConfigure {

    @Autowired
    private KafkaService kafkaService;

    @Autowired
    private KafkaConfigParam kafkaConfigParam;

   @Scheduled(cron = "0/10 * * * * ?")
    public void startListener() {
        List<String> topics = kafkaConfigParam.getStartTopics();
        System.out.println("开启。。。"+topics);
        Optional.ofNullable(topics).orElse(new ArrayList<>()).forEach(topic -> {
            kafkaService.start(topic);
        });

    }

    @Scheduled(cron = "0/10 * * * * ?")
    public void pauseListener() {
        List<String> topics = kafkaConfigParam.getPauseTopics();
        System.out.println("暂停。。。"+topics);
        Optional.ofNullable(topics).orElse(new ArrayList<>()).forEach(topic -> {
            kafkaService.pause(topic);
        });

    }
}

2.通过访问接口手动触发kafka消费的启动，暂停，恢复

 @RequestMapping("/start/{kafkaId}")
    public String start(@PathVariable String kafkaId) {
        if(!registry.getListenerContainer(kafkaId).isRunning()) {
            registry.getListenerContainer(kafkaId).start();
        } else {
            registry.getListenerContainer(kafkaId).resume();
        }

        return "ok";
    }

    @RequestMapping("/pause/{kafkaId}")
    public String pause(@PathVariable String kafkaId) {
        registry.getListenerContainer(kafkaId).pause();
        return "ok";
    }

3.监听nacos配置文件，完成动态的启停操作

/**
 * nacos配置变更监听
 * @author liangxi.zeng
 */
@Component
@Slf4j
public class NacosConfigListener {

    @Autowired
    private NacosConfigManager nacosConfigManager;

    @Autowired
    private KafkaService kafkaService;

    @Autowired
    private KafkaStartPauseParam kafkaStartPauseParam;

    /**
     * 分隔符
     */
    private static final String SPLIT=",";

    private static final String GROUP = "DEFAULT_GROUP";


    /**
     * nacos 配置文件监听
     * @throws NacosException
     */
    @PostConstruct
    private void reloadConfig() throws NacosException {
        nacosConfigManager.getConfigService().addListener(kafkaStartPauseParam.getDataId(), GROUP, new AbstractConfigChangeListener() {
            @Override
            public void receiveConfigChange(final ConfigChangeEvent event) {
                ConfigChangeItem pauseListeners = event.getChangeItem(KafkaStartPauseParam.PREFIX+".pause-listener");
                ConfigChangeItem startListeners = event.getChangeItem(KafkaStartPauseParam.PREFIX+".start-listener");
                if(Objects.nonNull(pauseListeners)) {
                    pause(pauseListeners);
                }

                if(Objects.nonNull(startListeners)) {
                    start(startListeners);
                }
            }
        });
    }

    /**
     * 暂停消费
     * @param pauseListeners
     */
    private void pause(ConfigChangeItem pauseListeners) {
        String pauseValue = pauseListeners.getNewValue();
        log.info("暂停listener:{}",pauseValue);
        if(!StringUtils.isEmpty(pauseValue)) {
            String[] pauseListenerIds = pauseValue.split(SPLIT);
            for(String pauseListenerId:pauseListenerIds) {
                kafkaService.pause(pauseListenerId);
            }
        }
    }

    /**
     * 恢复消费
     * @param startListeners
     */
    private void start(ConfigChangeItem startListeners) {
        String startValue = startListeners.getNewValue();
        log.info("启动listener:{}",startValue);
        if(!StringUtils.isEmpty(startValue)) {
            String[] startListenerIds = startValue.split(SPLIT);
            for(String startListenerId:startListenerIds) {
                kafkaService.start(startListenerId);
            }
        }
    }

}

配置类

/**
 * kafka配置参数
 * @author liangxi.zeng
 */
@ConfigurationProperties(prefix = KafkaStartPauseParam.PREFIX)
@Data
@Component
@RefreshScope
public class KafkaStartPauseParam {

    public static final String PREFIX = "tcl.kafka";

    private String pauseListener;

    private String startListener;

    private String dataId;
}

源码分析

1.springboot集成kafka,集成配置类org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration

2.@Import({KafkaAnnotationDrivenConfiguration.class})

 @Configuration(
        proxyBeanMethods = false
    )
    @EnableKafka
    @ConditionalOnMissingBean(
        name = {"org.springframework.kafka.config.internalKafkaListenerAnnotationProcessor"}
    )
    static class EnableKafkaConfiguration {
        EnableKafkaConfiguration() {
        }
    }

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(KafkaListenerConfigurationSelector.class)
public @interface EnableKafka {
}

@Order
public class KafkaListenerConfigurationSelector implements DeferredImportSelector {

    @Override
    public String[] selectImports(AnnotationMetadata importingClassMetadata) {
        return new String[] { KafkaBootstrapConfiguration.class.getName() };
    }

}

public class KafkaBootstrapConfiguration implements ImportBeanDefinitionRegistrar {

    @Override
    public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
        if (!registry.containsBeanDefinition(
                KafkaListenerConfigUtils.KAFKA_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)) {

            registry.registerBeanDefinition(KafkaListenerConfigUtils.KAFKA_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME,
                    new RootBeanDefinition(KafkaListenerAnnotationBeanPostProcessor.class));
        }

        if (!registry.containsBeanDefinition(KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)) {
            registry.registerBeanDefinition(KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
                    new RootBeanDefinition(KafkaListenerEndpointRegistry.class));
        }
    }

}

3.KafkaListenerAnnotationBeanPostProcessor这个类，就是消费监听的解析类，在此类中，将监听的方法放入了监听容器MessageListenerContainer

4.监听容器中有ListenerConsumer监听消费者的属性，此内部内实现了SchedulingAwareRunnable接口，此接口继承了Runnable接口，完成了定时异步消费等操作

@Override
        public void run() {
            
            while (isRunning()) {
                try {
                    pollAndInvoke();
                }
            }
            wrapUp();
        }

protected void pollAndInvoke() {
            if (!this.autoCommit && !this.isRecordAck) {
                processCommits();
            }
            idleBetweenPollIfNecessary();
            if (this.seeks.size() > 0) {
                processSeeks();
            }
            pauseConsumerIfNecessary();
            this.lastPoll = System.currentTimeMillis();
            this.polling.set(true);
            ConsumerRecords<K, V> records = doPoll();
            if (!this.polling.compareAndSet(true, false) && records != null) {
                /*
                 * There is a small race condition where wakeIfNecessary was called between
                 * exiting the poll and before we reset the boolean.
                 */
                if (records.count() > 0) {
                    this.logger.debug(() -> "Discarding polled records, container stopped: " + records.count());
                }
                return;
            }
            resumeConsumerIfNeccessary();
            debugRecords(records);
            if (records != null && records.count() > 0) {
                if (this.containerProperties.getIdleEventInterval() != null) {
                    this.lastReceive = System.currentTimeMillis();
                }
                invokeListener(records);
            }
            else {
                checkIdle();
            }
        }

遗留问题

在对kafka消费监听启停的过程中，发现当暂停消费的时候，对于存量的topic还是会消费完，不会立即停止，只是对于新产生的topic不会再消费了

源码地址

kafka消费热启停组件下载