技術文章 » 通過Nacos設定重新整理進行RabbitMQ消費者線上啟停

通過Nacos設定重新整理進行RabbitMQ消費者線上啟停

2023-02-13 06:01:14

前提

公司在做一些金融相關業務，某些時候由於資料提供商定期維護或者特殊原因需要暫停某些服務的消費者。之前選用的訊息佇列技術棧是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）