前提
公司在做一些金融相关业务,某些时候由于数据提供商定期维护或者特殊原因需要暂停某些服务的消费者。之前选用的消息队列技术栈是RabbitMQ
,用于微服务之间的消息投递,对于这类需要暂停消费者的场景是选用注释掉消费者Bean
中的相应Spring(Boot)
注解重新发布来实现,后面需要重新启动消费就是解开对应的注释再发布一次。这样的处理流程既繁琐,也显得没有技术含量,所以笔者就这个问题结合已有的配置中心Nacos
集群做了一个方案,使用Nacos
的配置准实时刷新功能去控制某个微服务实例的所有RabbitMQ
消费者(容器)的停止和启动。
方案原理
下面探讨一下方案的原理和可行性,主要包括:
- RabbitMQ消费者生命周期管理
- Nacos长轮询与配置刷新
因为工作中的主要技术栈是SpringBoot
+ RabbitMQ
,下文是探讨场景针对spring-boot-starter-amqp
(下面简称amqp
)展开。
使用SpringBoot版本为2.3.0.RELEASE,spring-cloud-alibaba-nacos-config的版本为2.2.0.RELEASE
RabbitMQ消费者生命周期管理
查看RabbitAnnotationDrivenConfiguration
的源码:
amqp
中默认启用spring.rabbitmq.listener.type=simple
,使用的RabbitListenerContainerFactory
(消息监听器容器工厂)实现为SimpleRabbitListenerContainerFactory
,使用的MessageListenerContainer
(消息监听器容器)实现为SimpleMessageListenerContainer
。在amqp
中,无论注解声明式或者编程式注册的消费者最终都会封装为MessageListenerContainer
实例,因此消费者生命周期可以直接通过MessageListenerContainer
进行管理,MessageListenerContainer
的生命周期管理API
会直接作用于最底层的真实消费者实现BlockingQueueConsumer
。几者的关系如下:
一般声明式消费者注册方式如下:
@Slf4j
@RabbitListener(id = "SingleAnnoMethodDemoConsumer", queues = "srd->srd.demo")
@Component
public class SingleAnnoMethodDemoConsumer {
@RabbitHandler
public void onMessage(Message message) {
log.info("SingleAnnoMethodDemoConsumer.onMessage => {}", new String(message.getBody(), StandardCharsets.UTF_8));
}
}
@RabbitListener(id = "MultiAnnoMethodDemoConsumer", queues = "srd->srd.demo")
@Component
@Slf4j
public class MultiAnnoMethodDemoConsumer {
@RabbitHandler
public void firstOnMessage(Message message) {
log.info("MultiAnnoMethodDemoConsumer.firstOnMessage => {}", new String(message.getBody(), StandardCharsets.UTF_8));
}
@RabbitHandler
public void secondOnMessage(Message message) {
log.info("MultiAnnoMethodDemoConsumer.secondOnMessage => {}", new String(message.getBody(), StandardCharsets.UTF_8));
}
}
@Component
@Slf4j
public class MultiAnnoInstanceDemoConsumer {
@RabbitListener(id = "MultiAnnoInstanceDemoConsumer-firstOnInstanceMessage", queues = "srd->srd.demo")
public void firstOnInstanceMessage(Message message) {
log.info("MultiAnnoInstanceDemoConsumer.firstOnInstanceMessage => {}", new String(message.getBody(), StandardCharsets.UTF_8));
}
@RabbitListener(id = "MultiAnnoInstanceDemoConsumer-secondOnInstanceMessage", queues = "srd->srd.sec")
public void secondOnInstanceMessage(Message message) {
log.info("MultiAnnoInstanceDemoConsumer.secondOnInstanceMessage => {}", new String(message.getBody(), StandardCharsets.UTF_8));
}
}
对于基于@RabbitListener
进行声明式注册的消费者,每个被@RabbitListener
修饰的Bean
或者方法最终都会单独生成一个SimpleMessageListenerContainer
实例,这些SimpleMessageListenerContainer
实例的唯一标识由@RabbitListener
的id
属性指定,缺省值为org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#N
,建议在使用时候通过规范约束必须定义此id
属性。分析源码可以得知这类型的消费者通过RabbitListenerAnnotationBeanPostProcessor
进行发现和自动注册,并且在RabbitListenerEndpointRegistry
缓存了注册信息,因此可以通过RabbitListenerEndpointRegistry
直接获取这些声明式的消费者容器实例:
RabbitListenerEndpointRegistry endpointRegistry = configurableListableBeanFactory.getBean(
RabbitListenerConfigUtils.RABBIT_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
RabbitListenerEndpointRegistry.class);
Set<String> listenerContainerIds = endpointRegistry.getListenerContainerIds();
for (String containerId : listenerContainerIds) {
MessageListenerContainer messageListenerContainer = endpointRegistry.getListenerContainer(containerId);
// do something with messageListenerContainer
}
一般编程式消费者注册方式如下:
// MessageListenerDemoConsumer
@Component
@Slf4j
public class MessageListenerDemoConsumer implements MessageListener {
@Override
public void onMessage(Message message) {
log.info("MessageListenerDemoConsumer.onMessage => {}", new String(message.getBody(), StandardCharsets.UTF_8));
}
}
// CustomMethodDemoConsumer
@Component
@Slf4j
public class CustomMethodDemoConsumer {
public void customOnMessage(Message message) {
log.info("CustomMethodDemoConsumer.customOnMessage => {}", new String(message.getBody(), StandardCharsets.UTF_8));
}
}
// configuration class
// 通过现存的MessageListener实例进行消费
@Bean
public SimpleMessageListenerContainer messageListenerDemoConsumerContainer(
ConnectionFactory connectionFactory,
@Qualifier("messageListenerDemoConsumer") MessageListener messageListener) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setListenerId("MessageListenerDemoConsumer");
container.setConnectionFactory(connectionFactory);
container.setConcurrentConsumers(1);
container.setMaxConcurrentConsumers(1);
container.setQueueNames("srd->srd.demo");
container.setAcknowledgeMode(AcknowledgeMode.AUTO);
container.setPrefetchCount(10);
container.setAutoStartup(true);
container.setMessageListener(messageListener);
return container;
}
// 通过IOC容器中某个Bean的具体方法进行消费
@Bean
public SimpleMessageListenerContainer customMethodDemoConsumerContainer(
ConnectionFactory connectionFactory,
CustomMethodDemoConsumer customMethodDemoConsumer) {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setListenerId("CustomMethodDemoConsumer");
container.setConnectionFactory(connectionFactory);
container.setConcurrentConsumers(1);
container.setMaxConcurrentConsumers(1);
container.setQueueNames("srd->srd.demo");
container.setAcknowledgeMode(AcknowledgeMode.AUTO);
container.setPrefetchCount(10);
container.setAutoStartup(true);
MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter();
messageListenerAdapter.setDelegate(customMethodDemoConsumer);
messageListenerAdapter.setDefaultListenerMethod("customOnMessage");
container.setMessageListener(messageListenerAdapter);
return container;
}
编程式注册的SimpleMessageListenerContainer
可以直接从IOC
容器中获取:
Map<String, MessageListenerContainer> messageListenerContainerBeans
= configurableListableBeanFactory.getBeansOfType(MessageListenerContainer.class);
if (!CollectionUtils.isEmpty(messageListenerContainerBeans)) {
messageListenerContainerBeans.forEach((beanId, messageListenerContainer) -> {
// do something with messageListenerContainer
});
}
至此,我们知道可以比较轻松地拿到服务中所有的MessageListenerContainer
的实例,从而可以管理服务内所有消费者的生命周期。
Nacos长轮询与配置刷新
Nacos
的客户端通过LongPolling
(长轮询)的方式监听Nacos
服务端集群对应dataId
和group
的配置数据变更,具体可以参考ClientWorker
的源码实现,实现的过程大致如下:
在非Spring(Boot)
体系中,可以通过ConfigService#addListener()
进行配置变更监听,示例代码如下:
Properties properties = new Properties();
properties.put(PropertyKeyConst.SERVER_ADDR, "127.0.0.1:8848");
properties.put(PropertyKeyConst.NAMESPACE, "LOCAL");
ConfigService configService = NacosFactory.createConfigService(properties);
Executor executor = Executors.newSingleThreadExecutor(runnable -> {
Thread thread = new Thread(runnable);
thread.setDaemon(true);
thread.setName("NacosConfigSyncWorker");
return thread;
});
configService.addListener("application-aplha.properties", "customer-service", new Listener() {
@Override
public Executor getExecutor() {
return executor;
}
@Override
public void receiveConfigInfo(String configInfo) {
// do something with 'configInfo'
}
});
这种LongPolling
的方式目前来看可靠性是比较高,因为Nacos
服务端集群一般在生产部署是大于3
的奇数个实例节点,并且底层基于raft
共识算法实现集群通讯,只要不是同一时间超过半数节点宕机集群还是能正常提供服务。但是从实现上来看会有一些局限性:
- 如果注册过多的配置变更监听器有可能会对
Nacos
服务端造成比较大的压力,毕竟是多个客户端进行轮询 - 配置变更是由
Nacos
客户端向Nacos
服务端发起请求,因此监听器回调有可能不是实时的(有可能延迟到客户端下一轮的LongPolling
提交) Nacos
客户端会缓存每次从Nacos
服务端拉取的配置内容,如果要变更配置文件过大有可能导致缓存的数据占用大量内存,影响客户端所在服务的性能
关于配置变更监听其实有其他候选的方案,例如Redis的发布订阅,Zookeeper的节点路径变更监听甚至是使用消息队列进行通知,本文使用Nacos配置变更监听的原因是更好的划分不同应用配置文件的编辑查看权限方便进行管理,其他候选方案要实现分权限管理需要二次开发
使用SpringCloudAlibaba
提供的spring-cloud-alibaba-nacos-config
可以更加简便地使用Nacos
配置刷新监听,并且会把变更的PropertySource
重新绑定到对应的配置属性Bean
。引入依赖:
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-alibaba-nacos-config</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba.nacos</groupId>
<artifactId>nacos-client</artifactId>
</dependency>
具体的配置类是NacosConfigProperties
:
红圈中是需要关注的配置项,refreshEnabled
是配置刷新的开关,默认是开启的。sharedConfigs
和extensionConfigs
虽然命名不同,但是两者实现和功能没有差异,都是类似于共享或者说扩展配置,每个共享(扩展)配置支持单独配置刷新开关。举个例子,在Nacos
服务端的某个配置如下图:
为了支持配置变更和对应的实体类成员变量更新,对应客户端的配置文件是这样的:
spring.cloud.nacos.config.refresh-enabled=true
spring.cloud.nacos.config.shared-configs[0].data-id=shared.properties
spring.cloud.nacos.config.shared-configs[0].group=shared-conf
spring.cloud.nacos.config.shared-configs[0].refresh=true
对应的配置属性Bean
如下:
@Data
@ConfigurationProperties(prefix = "shared")
public class SharedProperties {
private String foo;
}
只要客户端所在SpringBoot
服务启动完成后,修改Nacos
服务端对应dataId
为shared.properties
的shared.foo
属性值,那边SharedProperties
的foo
属性就会准实时刷新。可以在SharedProperties
添加一个@PostConstruct
来观察这个属性更新的过程:
@Slf4j
@Data
@ConfigurationProperties(prefix = "shared")
public class SharedProperties {
private final AtomicBoolean firstInit = new AtomicBoolean();
private String foo;
@PostConstruct
public void postConstruct() {
if (!firstInit.compareAndSet(false, true)) {
log.info("SharedProperties refresh...");
} else {
log.info("SharedProperties first init...");
}
}
}
方案实施
整个方案实施包括下面几步:
- 配置变更通知与配置类刷新
- 发现所有消费者容器
- 管理消费者容器生命周期
初始化一个Maven
项目,引入下面的依赖:
org.projectlombok:lombok:1.18.12
org.springframework.boot:spring-boot-starter-web:2.3.0.RELEASE
org.springframework.boot:spring-boot-starter-amqp:2.3.0.RELEASE
com.alibaba.cloud:spring-cloud-alibaba-nacos-config:2.2.0.RELEASE
com.alibaba.nacos:nacos-client:1.4.4
下载Nacos
服务并且启动一个单机实例(当前2023-02
的最新稳定版为2.2.0
),新建命名空间LOCAL
并且添加四份配置文件:
可以使用1.x的Nacos客户端去连接2.x的Nacos服务端,这个是Nacos做的向下兼容,反过来不行
前文提到的Nacos
客户端中,ConfigService
是通过dataId
和group
定位到具体的配置文件,一般dataId
按照配置文件的内容命名,对于SpringBoot
的应用配置文件一般命名为application-${profile}.[properties,yml]
,group
是配置文件的分组,对于SpringBoot
的应用配置文件一般命名为${spring.application.name}
。笔者在在这份SpringBoot
的应用配置文件中只添加了RabbitMQ
的配置:
确保本地或者远程有一个可用的RabbitMQ
服务,接下来往下开始实施方案。
配置变更通知与配置类刷新
前面已经提到过SpringBoot
结合Nacos
进行配置属性Bean
的成员变量刷新,在项目的Classpath
(resources
文件夹)添加bootstrap.properties
文件,内容如下:
spring.application.name=rabbitmq-rocketmq-demo
spring.profiles.active=default
# nacos配置
spring.cloud.nacos.config.enabled=true
spring.cloud.nacos.config.server-addr=127.0.0.1:8848
spring.cloud.nacos.config.namespace=LOCAL
spring.cloud.nacos.config.group=rabbitmq-rocketmq-demo
spring.cloud.nacos.config.prefix=application
spring.cloud.nacos.config.file-extension=properties
spring.cloud.nacos.config.refresh-enabled=true
spring.cloud.nacos.config.shared-configs[0].data-id=shared.properties
spring.cloud.nacos.config.shared-configs[0].group=shared-conf
spring.cloud.nacos.config.shared-configs[0].refresh=true
spring.cloud.nacos.config.extension-configs[0].data-id=extension.properties
spring.cloud.nacos.config.extension-configs[0].group=extension-conf
spring.cloud.nacos.config.extension-configs[0].refresh=true
spring.cloud.nacos.config.extension-configs[1].data-id=rabbitmq-toggle.properties
spring.cloud.nacos.config.extension-configs[1].group=rabbitmq-rocketmq-demo
spring.cloud.nacos.config.extension-configs[1].refresh=true
这里profile
定义为default
也就是会关联到Nacos
中dataId = 'application.properties', group = 'rabbitmq-rocketmq-demo'
那份配置文件,主要是用于定义amqp
需要的配置属性。对于RabbitMQ
消费者的开关,定义在dataId = 'rabbitmq-toggle.properties', group = 'rabbitmq-rocketmq-demo'
的文件中。添加RabbitmqToggleProperties
:
// RabbitmqToggleProperties
@Slf4j
@Data
@ConfigurationProperties(prefix = "rabbitmq.toggle")
public class RabbitmqToggleProperties {
private final AtomicBoolean firstInit = new AtomicBoolean();
private List<RabbitmqConsumer> consumers;
@PostConstruct
public void postConstruct() {
if (!firstInit.compareAndSet(false, true)) {
StaticEventPublisher.publishEvent(new RabbitmqToggleRefreshEvent(this));
log.info("RabbitmqToggleProperties refresh, publish RabbitmqToggleRefreshEvent...");
} else {
log.info("RabbitmqToggleProperties first init...");
}
}
@Data
public static class RabbitmqConsumer {
private String listenerId;
private Integer concurrentConsumers;
private Integer maxConcurrentConsumers;
private Boolean enable;
}
}
// RabbitmqToggleRefreshEvent
@Getter
public class RabbitmqToggleRefreshEvent extends ApplicationEvent {
private final RabbitmqToggleProperties rabbitmqToggleProperties;
public RabbitmqToggleRefreshEvent(RabbitmqToggleProperties rabbitmqToggleProperties) {
super("RabbitmqToggleRefreshEvent");
this.rabbitmqToggleProperties = rabbitmqToggleProperties;
}
}
// StaticEventPublisher
public class StaticEventPublisher {
private static ApplicationEventPublisher PUBLISHER = null;
public static void publishEvent(ApplicationEvent applicationEvent) {
if (Objects.nonNull(PUBLISHER)) {
PUBLISHER.publishEvent(applicationEvent);
}
}
public static void attachApplicationEventPublisher(ApplicationEventPublisher publisher) {
PUBLISHER = publisher;
}
}
这里prefix
定义为rabbitmq.toggle
,为了和rabbitmq-toggle.properties
的属性一一绑定,该文件中的配置Key
必须以rabbitmq.toggle
为前缀。RabbitmqToggleProperties
首次回调@PostConstruct
方法只打印初始化日志,再次回调@PostConstruct
方法则发布RabbitmqToggleRefreshEvent
事件,用于后面通知对应的消费者容器Bean
进行启停。
发现所有消费者容器
为了统一管理服务中所有消费者容器Bean
,需要定义一个类似于消费者容器注册或者缓存中心类,缓存Key
可以考虑使用listenerId
,Value
就直接使用MessageListenerContainer
实例即可:
private final ConcurrentMap<String, MessageListenerContainer> containerCache = Maps.newConcurrentMap();
这里既然选定了listenerId作为缓存的Key,那么必须定义好规范,要求无论注解声明式定义的消费者还是编程式定义的消费者,必须明确指定具体意义的listenerId,否则到时候存在Key的格式为org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#N会比较混乱
接下来发现和缓存所有消费者容器:
private ConfigurableListableBeanFactory configurableListableBeanFactory;
private ApplicationEventPublisher applicationEventPublisher;
// ----------------------------------------------------------------------
// 获取声明式消费者容器
RabbitListenerEndpointRegistry endpointRegistry = configurableListableBeanFactory.getBean(
RabbitListenerConfigUtils.RABBIT_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME,
RabbitListenerEndpointRegistry.class);
Set<String> listenerContainerIds = endpointRegistry.getListenerContainerIds();
for (String containerId : listenerContainerIds) {
MessageListenerContainer messageListenerContainer = endpointRegistry.getListenerContainer(containerId);
containerCache.putIfAbsent(containerId, messageListenerContainer);
}
// 获取编程式消费者容器
Map<String, MessageListenerContainer> messageListenerContainerBeans
= configurableListableBeanFactory.getBeansOfType(MessageListenerContainer.class);
if (!CollectionUtils.isEmpty(messageListenerContainerBeans)) {
messageListenerContainerBeans.forEach((beanId, bean) -> {
if (bean instanceof AbstractMessageListenerContainer) {
AbstractMessageListenerContainer abstractMessageListenerContainer = (AbstractMessageListenerContainer) bean;
String listenerId = abstractMessageListenerContainer.getListenerId();
if (StringUtils.hasLength(listenerId)) {
containerCache.putIfAbsent(listenerId, abstractMessageListenerContainer);
} else {
containerCache.putIfAbsent(beanId, bean);
}
} else {
containerCache.putIfAbsent(beanId, bean);
}
});
}
Set<String> listenerIds = containerCache.keySet();
listenerIds.forEach(listenerId -> log.info("Cache message listener container => {}", listenerId));
// 所有消费者容器Bean发现完成后才接收刷新事件
StaticEventPublisher.attachApplicationEventPublisher(this.applicationEventPublisher);
StaticEventPublisher
中的ApplicationEventPublisher
属性延迟到所有消费者容器缓存完成后赋值,防止过早的属性变更通知导致部分消费者容器的启停操作被忽略。
管理消费者容器生命周期
接收到RabbitmqToggleRefreshEvent
事件后,然后遍历传递过来的RabbitmqToggleProperties
里面的consumers
,再基于已经发现的消费者容器进行处理,代码大概如下:
@EventListener(classes = RabbitmqToggleRefreshEvent.class)
public void onRabbitmqToggleRefreshEvent(RabbitmqToggleRefreshEvent event) {
RabbitmqToggleProperties rabbitmqToggleProperties = event.getRabbitmqToggleProperties();
List<RabbitmqToggleProperties.RabbitmqConsumer> consumers = rabbitmqToggleProperties.getConsumers();
if (!CollectionUtils.isEmpty(consumers)) {
consumers.forEach(consumerConf -> {
String listenerId = consumerConf.getListenerId();
if (StringUtils.hasLength(listenerId)) {
MessageListenerContainer messageListenerContainer = containerCache.get(listenerId);
if (Objects.nonNull(messageListenerContainer)) {
// running -> stop
if (messageListenerContainer.isRunning() && Objects.equals(Boolean.FALSE, consumerConf.getEnable())) {
messageListenerContainer.stop();
log.info("Message listener container => {} stop successfully", listenerId);
}
// modify concurrency
if (messageListenerContainer instanceof SimpleMessageListenerContainer) {
SimpleMessageListenerContainer simpleMessageListenerContainer
= (SimpleMessageListenerContainer) messageListenerContainer;
if (Objects.nonNull(consumerConf.getConcurrentConsumers())) {
simpleMessageListenerContainer.setConcurrentConsumers(consumerConf.getConcurrentConsumers());
}
if (Objects.nonNull(consumerConf.getMaxConcurrentConsumers())) {
simpleMessageListenerContainer.setMaxConcurrentConsumers(consumerConf.getMaxConcurrentConsumers());
}
}
// stop -> running
if (!messageListenerContainer.isRunning() && Objects.equals(Boolean.TRUE, consumerConf.getEnable())) {
messageListenerContainer.start();
log.info("Message listener container => {} start successfully", listenerId);
}
}
}
});
}
}
修改Nacos
服务里面的rabbitmq-toggle.properties
文件,输入内容如下:
rabbitmq.toggle.consumers[0].listenerId=MultiAnnoInstanceDemoConsumer-firstOnInstanceMessage
rabbitmq.toggle.consumers[0].enable=true
rabbitmq.toggle.consumers[1].listenerId=MultiAnnoInstanceDemoConsumer-secondOnInstanceMessage
rabbitmq.toggle.consumers[1].enable=true
rabbitmq.toggle.consumers[2].listenerId=MultiAnnoMethodDemoConsumer
rabbitmq.toggle.consumers[2].enable=true
rabbitmq.toggle.consumers[3].listenerId=SingleAnnoMethodDemoConsumer
rabbitmq.toggle.consumers[3].enable=true
rabbitmq.toggle.consumers[4].listenerId=CustomMethodDemoConsumer
rabbitmq.toggle.consumers[4].enable=true
rabbitmq.toggle.consumers[5].listenerId=MessageListenerDemoConsumer
rabbitmq.toggle.consumers[5].enable=true
启动项目,观察RabbitMQ WebUI
对应的队列消费者数量:
然后随机修改rabbitmq-toggle.properties
文件某个消费者容器设置为enable = 'fasle'
,观察服务日志和观察RabbitMQ WebUI
的变化:
可见RabbitMQ WebUI
中队列消费者数量减少,服务日志也提示listenerId = 'MessageListenerDemoConsumer'
的消费者容器被停止了。
一些思考
为了更精确控制有消费者容器的启停,可以考虑在配置文件中定义关闭消费者容器的自动启动开关:
spring.rabbitmq.listener.simple.auto-startup=false
可以考虑在RabbitmqToggleProperties
首次回调@PostConstruct
方法时候发布RabbitmqToggleInitEvent
事件,然后监听此事件启动所有已经发现的消费者容器。这样就能做到应用内部的消费者的启停行为总是以Nacos
的开关配置文件为准,并且可以实现在线启停和动态调整最小最大消费者数量。
另外,如果细心的话能够观察到服务日志中,每当监听到Nacos
配置变动会打印Started application in N seconds (JVM running for M)
的日志,这个并不是服务重启了,而是启动了一个Spring
子容器用于构建一个全新的StandardEnvironment
(见文末Demo
项目中的EnvironmentCaptureApplicationRunner
)用来承载刷新后的配置文件内容,然后再拷贝或者覆盖到当前的Spring
容器中的PropertySources
,这个过程的代码实现类似这样:
小结
本文探讨了一种通过Nacos
配置刷新方式管理SpringBoot
服务中RabbitMQ
消费者生命周期管理的方案,目前只是提供了完整的思路和一些Demo
级别代码,后续应该会完善方案和具体的工程级别编码实现。
本文Demo
项目仓库:
(本文完 c-3-d e-a-20230212)
标签:spring,Nacos,rabbitmq,nacos,启停,RabbitMQ,public,cloud From: https://www.cnblogs.com/throwable/p/17114740.html