项目场景:
在月结,或者某些时候,我们需要停掉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不会再消费了文章来源:https://www.toymoban.com/news/detail-459080.html
源码地址
kafka消费热启停组件下载文章来源地址https://www.toymoban.com/news/detail-459080.html
到了这里,关于springboot集成kafka消费手动启动停止的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!