项目场景： 在月结，或者某些时候，我们需要停掉kafka所有的消费端，让其暂时停止消费，而后等月结完成，再从新对消费监听恢复，进行消费，此动作不需要重启服务 解决分析 Kafk
项目场景：
在月结，或者某些时候，我们需要停掉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不会再消费了
到此这篇关于springboot集成kafka消费手动启动停止的文章就介绍到这了,更多相关springboot集成kafka内容请搜索自由互联以前的文章或继续浏览下面的相关文章希望大家以后多多支持自由互联！