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

    项目场景：

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

    解决分析

    KafkaListenerEndpointRegistry这是kafka与spring集成的监听注册bean，可以通过它获取监听容器对象，然后对监听容器对象实行启动，暂停，恢复等操作
    1. /**
    2.  * kafka服务操作类
    3.  * @author liangxi.zeng
    4.  */
    5. @Service
    6. @Slf4j
    7. public class KafkaService {
    8.  
    9.     @Autowired
    10.     private KafkaListenerEndpointRegistry registry;
    11.  
    12.     /**
    13.      * 开启消费
    14.      * @param listenerId
    15.      */
    16.     public void start(String listenerId) {
    17.         MessageListenerContainer messageListenerContainer = registry
    18.                 .getListenerContainer(listenerId);
    19.         if(Objects.nonNull(messageListenerContainer)) {
    20.             if(!messageListenerContainer.isRunning()) {
    21.                 messageListenerContainer.start();
    22.             } else {
    23.                 if(messageListenerContainer.isContainerPaused()) {
    24.                     log.info("listenerId:{},恢复",listenerId);
    25.                     messageListenerContainer.resume();
    26.                 }
    27.             }
    28.         }
    29.     }
    30.  
    31.     /**
    32.      * 停止消费
    33.      * @param listenerId
    34.      */
    35.     public void pause(String listenerId) {
    36.         MessageListenerContainer messageListenerContainer = registry
    37.                 .getListenerContainer(listenerId);
    38.         if(Objects.nonNull(messageListenerContainer) && !messageListenerContainer.isContainerPaused()) {
    39.             log.info("listenerId:{},暂停",listenerId);
    40.             messageListenerContainer.pause();
    41.         }
    42.     }
    43. }

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

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

    1. /**
    2.  * 卡夫卡配置类
    3.  * @author liangxi.zeng
    4.  */
    5. @Configuration
    6. @EnableScheduling
    7. public class KafkaConfigure {
    8.  
    9.     @Autowired
    10.     private KafkaService kafkaService;
    11.  
    12.     @Autowired
    13.     private KafkaConfigParam kafkaConfigParam;
    14.  
    15.    @Scheduled(cron = "0/10 * * * * ?")
    16.     public void startListener() {
    17.         List topics = kafkaConfigParam.getStartTopics();
    18.         System.out.println("开启。。。"+topics);
    19.         Optional.ofNullable(topics).orElse(new ArrayList<>()).forEach(topic -> {
    20.             kafkaService.start(topic);
    21.         });
    22.  
    23.     }
    24.  
    25.     @Scheduled(cron = "0/10 * * * * ?")
    26.     public void pauseListener() {
    27.         List topics = kafkaConfigParam.getPauseTopics();
    28.         System.out.println("暂停。。。"+topics);
    29.         Optional.ofNullable(topics).orElse(new ArrayList<>()).forEach(topic -> {
    30.             kafkaService.pause(topic);
    31.         });
    32.  
    33.     }
    34. }

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

    1.  @RequestMapping("/start/{kafkaId}")
    2.     public String start(@PathVariable String kafkaId) {
    3.         if(!registry.getListenerContainer(kafkaId).isRunning()) {
    4.             registry.getListenerContainer(kafkaId).start();
    5.         } else {
    6.             registry.getListenerContainer(kafkaId).resume();
    7.         }
    8.  
    9.         return "ok";
    10.     }
    11.  
    12.     @RequestMapping("/pause/{kafkaId}")
    13.     public String pause(@PathVariable String kafkaId) {
    14.         registry.getListenerContainer(kafkaId).pause();
    15.         return "ok";
    16.     }

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

    1. /**
    2.  * nacos配置变更监听
    3.  * @author liangxi.zeng
    4.  */
    5. @Component
    6. @Slf4j
    7. public class NacosConfigListener {
    8.  
    9.     @Autowired
    10.     private NacosConfigManager nacosConfigManager;
    11.  
    12.     @Autowired
    13.     private KafkaService kafkaService;
    14.  
    15.     @Autowired
    16.     private KafkaStartPauseParam kafkaStartPauseParam;
    17.  
    18.     /**
    19.      * 分隔符
    20.      */
    21.     private static final String SPLIT=",";
    22.  
    23.     private static final String GROUP = "DEFAULT_GROUP";
    24.  
    25.  
    26.     /**
    27.      * nacos 配置文件监听
    28.      * @throws NacosException
    29.      */
    30.     @PostConstruct
    31.     private void reloadConfig() throws NacosException {
    32.         nacosConfigManager.getConfigService().addListener(kafkaStartPauseParam.getDataId(), GROUP, new AbstractConfigChangeListener() {
    33.             @Override
    34.             public void receiveConfigChange(final ConfigChangeEvent event) {
    35.                 ConfigChangeItem pauseListeners = event.getChangeItem(KafkaStartPauseParam.PREFIX+".pause-listener");
    36.                 ConfigChangeItem startListeners = event.getChangeItem(KafkaStartPauseParam.PREFIX+".start-listener");
    37.                 if(Objects.nonNull(pauseListeners)) {
    38.                     pause(pauseListeners);
    39.                 }
    40.  
    41.                 if(Objects.nonNull(startListeners)) {
    42.                     start(startListeners);
    43.                 }
    44.             }
    45.         });
    46.     }
    47.  
    48.     /**
    49.      * 暂停消费
    50.      * @param pauseListeners
    51.      */
    52.     private void pause(ConfigChangeItem pauseListeners) {
    53.         String pauseValue = pauseListeners.getNewValue();
    54.         log.info("暂停listener:{}",pauseValue);
    55.         if(!StringUtils.isEmpty(pauseValue)) {
    56.             String[] pauseListenerIds = pauseValue.split(SPLIT);
    57.             for(String pauseListenerId:pauseListenerIds) {
    58.                 kafkaService.pause(pauseListenerId);
    59.             }
    60.         }
    61.     }
    62.  
    63.     /**
    64.      * 恢复消费
    65.      * @param startListeners
    66.      */
    67.     private void start(ConfigChangeItem startListeners) {
    68.         String startValue = startListeners.getNewValue();
    69.         log.info("启动listener:{}",startValue);
    70.         if(!StringUtils.isEmpty(startValue)) {
    71.             String[] startListenerIds = startValue.split(SPLIT);
    72.             for(String startListenerId:startListenerIds) {
    73.                 kafkaService.start(startListenerId);
    74.             }
    75.         }
    76.     }
    77.  
    78. }

    配置类

    1. /**
    2.  * kafka配置参数
    3.  * @author liangxi.zeng
    4.  */
    5. @ConfigurationProperties(prefix = KafkaStartPauseParam.PREFIX)
    6. @Data
    7. @Component
    8. @RefreshScope
    9. public class KafkaStartPauseParam {
    10.  
    11.     public static final String PREFIX = "tcl.kafka";
    12.  
    13.     private String pauseListener;
    14.  
    15.     private String startListener;
    16.  
    17.     private String dataId;
    18. }

    源码分析

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

    2.@Import({KafkaAnnotationDrivenConfiguration.class})

    1.  @Configuration(
    2.         proxyBeanMethods = false
    3.     )
    4.     @EnableKafka
    5.     @ConditionalOnMissingBean(
    6.         name = {"org.springframework.kafka.config.internalKafkaListenerAnnotationProcessor"}
    7.     )
    8.     static class EnableKafkaConfiguration {
    9.         EnableKafkaConfiguration() {
    10.         }
    11.     }
    1. @Target(ElementType.TYPE)
    2. @Retention(RetentionPolicy.RUNTIME)
    3. @Documented
    4. @Import(KafkaListenerConfigurationSelector.class)
    5. public @interface EnableKafka {
    6. }
    1. @Order
    2. public class KafkaListenerConfigurationSelector implements DeferredImportSelector {
    3.  
    4. 	@Override
    5. 	public String[] selectImports(AnnotationMetadata importingClassMetadata) {
    6. 		return new String[] { KafkaBootstrapConfiguration.class.getName() };
    7. 	}
    8.  
    9. }
    1. public class KafkaBootstrapConfiguration implements ImportBeanDefinitionRegistrar {
    2.  
    3. 	@Override
    4. 	public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
    5. 		if (!registry.containsBeanDefinition(
    6. 				KafkaListenerConfigUtils.KAFKA_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME)) {
    7.  
    8. 			registry.registerBeanDefinition(KafkaListenerConfigUtils.KAFKA_LISTENER_ANNOTATION_PROCESSOR_BEAN_NAME,
    9. 					new RootBeanDefinition(KafkaListenerAnnotationBeanPostProcessor.class));
    10. 		}
    11.  
    12. 		if (!registry.containsBeanDefinition(KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)) {
    13. 			registry.registerBeanDefinition(KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
    14. 					new RootBeanDefinition(KafkaListenerEndpointRegistry.class));
    15. 		}
    16. 	}
    17.  
    18. }

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

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

    1. @Override
    2. 		public void run() {
    3. 			
    4. 			while (isRunning()) {
    5. 				try {
    6. 					pollAndInvoke();
    7. 				}
    8. 			}
    9. 			wrapUp();
    10. 		}
    11.  
    12. protected void pollAndInvoke() {
    13. 			if (!this.autoCommit && !this.isRecordAck) {
    14. 				processCommits();
    15. 			}
    16. 			idleBetweenPollIfNecessary();
    17. 			if (this.seeks.size() > 0) {
    18. 				processSeeks();
    19. 			}
    20. 			pauseConsumerIfNecessary();
    21. 			this.lastPoll = System.currentTimeMillis();
    22. 			this.polling.set(true);
    23. 			ConsumerRecords records = doPoll();
    24. 			if (!this.polling.compareAndSet(true, false) && records != null) {
    25. 				/*
    26. 				 * There is a small race condition where wakeIfNecessary was called between
    27. 				 * exiting the poll and before we reset the boolean.
    28. 				 */
    29. 				if (records.count() > 0) {
    30. 					this.logger.debug(() -> "Discarding polled records, container stopped: " + records.count());
    31. 				}
    32. 				return;
    33. 			}
    34. 			resumeConsumerIfNeccessary();
    35. 			debugRecords(records);
    36. 			if (records != null && records.count() > 0) {
    37. 				if (this.containerProperties.getIdleEventInterval() != null) {
    38. 					this.lastReceive = System.currentTimeMillis();
    39. 				}
    40. 				invokeListener(records);
    41. 			}
    42. 			else {
    43. 				checkIdle();
    44. 			}
    45. 		}

    遗留问题

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

    源码地址 

    kafka消费热启停组件下载

  • 相关阅读:
    react 中setState 的三种写法
    【开发技术】2万字详细介绍Docker 和 web项目的部署监控，docker部署，拉取kafana，prometheus镜像监控
    Pandas 操作数据（三）
    Linux学习记录——이십구 网络基础（2）
    洛谷_P1339 [USACO09OCT]Heat Wave G_最短路
    new`是如何创建对象实例的？
    Web前端：如何为Web应用程序开发找到合适的技术堆栈?
    Java“牵手”淘宝商品评论数据采集+淘宝商品评价接口，淘宝商品追评数据接口，行业商品质检接口，API接口申请指南
    Presto资源管理之Resource Groups And Selector
    pytorch训练错误记录
  • 原文地址：https://blog.csdn.net/zengliangxi/article/details/126711032