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

/**
 * 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();
        }
    }
}

/**
 * 卡夫卡配置类
 * @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);
        });

    }
}

@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";
    }

/**
 * 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;
}

@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));
		}
	}

}

@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();
			}
		}