项目场景：
在月结，或者某些时候，我们需要停掉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内容请搜索Devmax以前的文章或继续浏览下面的相关文章希望大家以后多多支持Devmax！