簡介 在開發(fā)中使用線程池去執(zhí)行異步任務是比較普遍的操作，然而雖然有些異步操作我們并不十分要求可靠性和實時性，但總歸業(yè)務還是需要的。如果在每次的服務發(fā)版過程中，我們不去介入線程池的停機邏輯，那么很有可能就會造成線程池中隊列的任務還未執(zhí)行完成，自然就會造成數(shù)據(jù)的丟失。

探究

注意，本文所有前提是對進程進行下線時使用的是kill -15

我們知道Spring已經(jīng)實現(xiàn)了自己的優(yōu)雅停機方案，詳細請參考org.springframework.context.support.AbstractApplicationContext#registerShutdownHook，然后主要看調(diào)用的org.springframework.context.support.AbstractApplicationContext#doClose， 在這個方法里定義了容器銷毀的執(zhí)行順序

protected void doClose() { // Check whether an actual close attempt is necessary… if (this.active.get() && this.closed.compareAndSet(false, true)) { if (logger.isDebugEnabled()) { logger.debug(“Closing ” + this); } LiveBeansView.unregisterApplicationContext(this); try { // Publish shutdown event. publishEvent(new ContextClosedEvent(this)); } catch (Throwable ex) { logger.warn(“Exception thrown from ApplicationListener handling ContextClosedEvent”, ex); } // Stop all Lifecycle beans, to avoid delays during inpidual destruction. if (this.lifecycleProcessor != null) { try { this.lifecycleProcessor.onClose(); } catch (Throwable ex) { logger.warn(“Exception thrown from LifecycleProcessor on context close”, ex); } } // Destroy all cached singletons in the context’s BeanFactory. destroyBeans(); // Close the state of this context itself. closeBeanFactory(); // Let subclasses do some final clean-up if they wish… onClose(); // Reset local application listeners to pre-refresh state. if (this.earlyApplicationListeners != null) { this.applicationListeners.clear(); this.applicationListeners.addAll(this.earlyApplicationListeners); } // Switch to inactive. this.active.set(false); } }

我們先主要關(guān)注下destroyBeans這個方法，看bean的銷毀邏輯是什么，然后看到了下面的一個bean的銷毀順序邏輯,具體方法在org.springframework.beans.factory.support.DefaultSingletonBeanRegistry#destroySingletons

private final Map disposableBeans = new LinkedHashMap(); public void destroySingletons() { if (logger.isTraceEnabled()) { logger.trace(“Destroying singletons in ” + this); } synchronized (this.singletonObjects) { this.singletonsCurrentlyInDestruction = true; } String[] disposableBeanNames; synchronized (this.disposableBeans) { disposableBeanNames = StringUtils.toStringArray(this.disposableBeans.keySet()); } for (int i = disposableBeanNames.length – 1; i >= 0; i–) { destroySingleton(disposableBeanNames[i]); } this.containedBeanMap.clear(); this.dependentBeanMap.clear(); this.dependenciesForBeanMap.clear(); clearSingletonCache(); }

可以看到最至關(guān)重要的就是一個屬性disposableBeans,這個屬性是一個LinkedHashMap， 因此屬性是有序的，所以銷毀的時候也是按照某種規(guī)則保持和放入一樣的順序進行銷毀的，現(xiàn)在就是要確認這個屬性里到底存的是什么。

經(jīng)過調(diào)試發(fā)現(xiàn)，在創(chuàng)建bean的org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory#doCreateBean方法中，會調(diào)用一個方法org.springframework.beans.factory.support.AbstractBeanFactory#registerDisposableBeanIfNecessary， 在這個方法中會調(diào)用org.springframework.beans.factory.support.DefaultSingletonBeanRegistry#registerDisposableBean然后將當前創(chuàng)建的bean放入到屬性disposableBeans中，那么現(xiàn)在來看一下放入的邏輯什么？

相關(guān)代碼貼一下

protected void registerDisposableBeanIfNecessary(String beanName, Object bean, RootBeanDefinition mbd) { AccessControlContext acc = (System.getSecurityManager() != null ? getAccessControlContext() : null); if (!mbd.isPrototype() && requiresDestruction(bean, mbd)) { if (mbd.isSingleton()) { // Register a DisposableBean implementation that performs all destruction // work for the given bean: DestructionAwareBeanPostProcessors, // DisposableBean interface, custom destroy method. registerDisposableBean(beanName, new DisposableBeanAdapter(bean, beanName, mbd, getBeanPostProcessors(), acc)); } else { // A bean with a custom scope… Scope scope = this.scopes.get(mbd.getScope()); if (scope == null) { throw new IllegalStateException(“No Scope registered for scope name ‘” + mbd.getScope() + “‘”); } scope.registerDestructionCallback(beanName, new DisposableBeanAdapter(bean, beanName, mbd, getBeanPostProcessors(), acc)); } }}

org.springframework.beans.factory.support.AbstractBeanFactory#requiresDestruction

protected boolean requiresDestruction(Object bean, RootBeanDefinition mbd) { return (bean != null && (DisposableBeanAdapter.hasDestroyMethod(bean, mbd) || (hasDestructionAwareBeanPostProcessors() && DisposableBeanAdapter.hasApplicableProcessors(bean, getBeanPostProcessors())))); }

經(jīng)過兩個方法可以看到如果一個bean的scope是singleton并且這個bean實現(xiàn)了org.springframework.beans.factory.DisposableBean這個接口的destroy()方法，那么就會滿足條件。

現(xiàn)在可以確定一點，如果我們將線程池交給Spring管理，并且實現(xiàn)它的close方法，就可以在應用收到下線信號的時候執(zhí)行這個bean的銷毀方法，那么我們就可以在銷毀方法中寫線程池的停機邏輯。

我們知道Spring提供了線程池的封裝，在Spring中如果我們要定義線程池一般會使用org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor以及用于任務調(diào)度的org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler，先來簡單看個定義ThreadPoolTaskExecutor線程池的例子

@Configurationpublic class ThreadConfig { @Bean public ThreadPoolTaskExecutor testExecutor() { ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor(); threadPoolTaskExecutor.setThreadNamePrefix(“test-shutdown-pool-“); threadPoolTaskExecutor.setCorePoolSize(1); threadPoolTaskExecutor.setMaxPoolSize(1); threadPoolTaskExecutor.setKeepAliveSeconds(60); threadPoolTaskExecutor.setQueueCapacity(1000); threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy()); return threadPoolTaskExecutor; }}

現(xiàn)在來一下線程池的這個類結(jié)構(gòu)，ThreadPoolTaskExecutor繼承了org.springframework.scheduling.concurrent.ExecutorConfigurationSupport, 實現(xiàn)了org.springframework.beans.factory.DisposableBean,完整結(jié)構(gòu)如下

public class ThreadPoolTaskExecutor extends ExecutorConfigurationSupport implements AsyncListenableTaskExecutor, SchedulingTaskExecutor { }public abstract class ExecutorConfigurationSupport extends CustomizableThreadFactory implements BeanNameAware, InitializingBean, DisposableBean {}

從這里就能看到其實線程池類ThreadPoolTaskExecutor是滿足最開始看到的銷毀條件的，那么現(xiàn)在就來看下在父類ExecutorConfigurationSupport中定義的destroy()方法,將其中關(guān)鍵部分代碼摘錄下來

public abstract class ExecutorConfigurationSupport extends CustomizableThreadFactory implements BeanNameAware, InitializingBean, DisposableBean { private boolean waitForTasksToCompleteOnShutdown = false; private long awaitTerminationMillis = 0; @Nullable private ExecutorService executor; @Override public void destroy() { shutdown(); } /** * Perform a shutdown on the underlying ExecutorService. * @see java.util.concurrent.ExecutorService#shutdown() * @see java.util.concurrent.ExecutorService#shutdownNow() */ public void shutdown() { if (logger.isInfoEnabled()) { logger.info(“Shutting down ExecutorService” + (this.beanName != null ? ” ‘” + this.beanName + “‘” : “”)); } if (this.executor != null) { if (this.waitForTasksToCompleteOnShutdown) { this.executor.shutdown(); } else { for (Runnable remainingTask : this.executor.shutdownNow()) { cancelRemainingTask(remainingTask); } } awaitTerminationIfNecessary(this.executor); } } private void awaitTerminationIfNecessary(ExecutorService executor) { if (this.awaitTerminationMillis > 0) { try { if (!executor.awaitTermination(this.awaitTerminationMillis, TimeUnit.MILLISECONDS)) { if (logger.isWarnEnabled()) { logger.warn(“Timed out while waiting for executor” + (this.beanName != null ? ” ‘” + this.beanName + “‘” : “”) + ” to terminate”); } } } catch (InterruptedException ex) { if (logger.isWarnEnabled()) { logger.warn(“Interrupted while waiting for executor” + (this.beanName != null ? ” ‘” + this.beanName + “‘” : “”) + ” to terminate”); } Thread.currentThread().interrupt(); } } } protected void cancelRemainingTask(Runnable task) { if (task instanceof Future) { ((Future) task).cancel(true); } }}

整個的邏輯還是比較清晰的， 在容器銷毀的時候會調(diào)用本地shutdown()方法， 在這個方法中會去判斷waitForTasksToCompleteOnShutdown這個的屬性，如果為true, 則調(diào)用線程池的shutdown()方法，這個方法并不會讓線程池立即停止，而是不再接受新的任務并繼續(xù)執(zhí)行已經(jīng)在隊列中的任務。如果為false, 則取消任務隊列中的剩余任務。而這個屬性的默認值為false。因此默認是不具備我們需要的功能的。

然而無論這個值的屬性最終是否為TRUE，最終都會調(diào)用方法awaitTerminationIfNecessary()， 線程的停止無論是shutdown還是shutdownNow都無法保證線程池能夠停止下來，因為需要配合線程池的方法awaitTermination使用，在這個方法中指定一個最大等待時間，則能夠保證線程池最終一定可以被停止下來。

不知道有沒有注意到一個細節(jié)，上述所有對線程池的操作使用的屬性都是private ExecutorService executor;，那么這個executor是什么時候賦值的呢？

畢竟我們在創(chuàng)建bean的時候是直接new的ThreadPoolTaskExecutor，并沒有去處理這個屬性。還是看線程池的父類ExecutorConfigurationSupport, 其實現(xiàn)了接口org.springframework.beans.factory.InitializingBean，在容器初始化完成后有這樣一段代碼

@Override public void afterPropertiesSet() { initialize(); } /** * Set up the ExecutorService. */ public void initialize() { if (logger.isInfoEnabled()) { logger.info(“Initializing ExecutorService” + (this.beanName != null ? ” ‘” + this.beanName + “‘” : “”)); } if (!this.threadNamePrefixSet && this.beanName != null) { setThreadNamePrefix(this.beanName + “-“); } this.executor = initializeExecutor(this.threadFactory, this.rejectedExecutionHandler); } protected abstract ExecutorService initializeExecutor( ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler);

線程池bean在初始化完成后會調(diào)用父類的afterPropertiesSet方法，上面的代碼已經(jīng)很清晰的說明了問題， 最終父類中又定義了抽象方法initializeExecutor()，供子類去具體實現(xiàn)如果初始化這個屬性executor， 因為我們知道線程池的實現(xiàn)除了普通的異步任務線程池ThreadPoolTaskExecutor, 還有基于定時調(diào)度的線程池ThreadPoolTaskExecutor， 具體實現(xiàn)這里就不貼出來了，反正已經(jīng)能夠看出來這個屬性是如何被賦值的了，所以上述銷毀時代碼可以直接使用。

現(xiàn)在整體總結(jié)下來，其實發(fā)現(xiàn)我們Spring已經(jīng)幫我們實現(xiàn)了線程池的優(yōu)雅停機規(guī)則，在接收到停機信號時，先拒絕接收新的任務，并繼續(xù)執(zhí)行已經(jīng)接受的任務，在任務執(zhí)行完成或者到達最大等待時間，完成線程池的關(guān)閉。這么一整套邏輯正是我們所需要的，而我們?nèi)绻褂眠@個邏輯，僅僅需要在配置線程池的時候指定下上面看到的waitForTasksToCompleteOnShutdown屬性和awaitTerminationMillis屬性。

修改一下上面之前寫的線程池定義代碼, 將waitForTasksToCompleteOnShutdown屬性設置為true, 并指定awaitTerminationMillis。

@Configurationpublic class ThreadConfig { @Bean public ThreadPoolTaskExecutor testExecutor() { ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor(); threadPoolTaskExecutor.setThreadNamePrefix(“test-shutdown-pool-“); threadPoolTaskExecutor.setCorePoolSize(1); threadPoolTaskExecutor.setMaxPoolSize(1); threadPoolTaskExecutor.setKeepAliveSeconds(60); threadPoolTaskExecutor.setQueueCapacity(1000); threadPoolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true); threadPoolTaskExecutor.setAwaitTerminationSeconds(60); threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy()); return threadPoolTaskExecutor; }}

那么到目前這樣處理是不是就沒有問題了呢？

要分情況來看， 如果按照上述操作，是能夠保證最初預期目標的。線程池會在接收到下線指令時停止接受新的任務，并繼續(xù)執(zhí)行隊列中的未完成的任務，直到任務完成或者達到指定的最大等待時間。

如果任務是一些不操作其它資源的，只是一些本地計算或者日志什么之類的，那么任務不會出問題。但是如果任務本身依賴各種數(shù)據(jù)源，比如數(shù)據(jù)庫、緩存等之類的，那么就會出現(xiàn)任務本身會執(zhí)行，但是卻會失敗的問題，因為數(shù)據(jù)源已經(jīng)早于線程池關(guān)閉了。

那么，能不能控制數(shù)據(jù)源和線程池的銷毀順序呢？在上面我們看到銷毀順序的時候看到了線程池會在放入到disposableBeans的原因，其實數(shù)據(jù)源也是會被放入到這個屬性中的，這個原因和Spring的生命周期無關(guān)，而是另外一個判斷條件。

之前沒有貼出來具體代碼，現(xiàn)在來看下org.springframework.beans.factory.support.AbstractBeanFactory#requiresDestruction方法中的調(diào)用的另外一個本地方法org.springframework.beans.factory.support.DisposableBeanAdapter#hasDestroyMethod

public static boolean hasDestroyMethod(Object bean, RootBeanDefinition beanDefinition) { if (bean instanceof DisposableBean || bean instanceof AutoCloseable) { return true; } String destroyMethodName = beanDefinition.getDestroyMethodName(); if (AbstractBeanDefinition.INFER_METHOD.equals(destroyMethodName)) { return (ClassUtils.hasMethod(bean.getClass(), CLOSE_METHOD_NAME) || ClassUtils.hasMethod(bean.getClass(), SHUTDOWN_METHOD_NAME)); } return StringUtils.hasLength(destroyMethodName); }

之前線程池能夠執(zhí)行銷毀流程是因為實現(xiàn)了接口DisposableBean， 而數(shù)據(jù)源則是實現(xiàn)了另外一個接口AutoCloseable， 因此數(shù)據(jù)源也是會執(zhí)行銷毀邏輯的。

現(xiàn)在我們只要介入bean的創(chuàng)建優(yōu)先級即可， 使用org.springframework.core.annotation.Order注解來指定線程池創(chuàng)建的高優(yōu)先級，如下。

@Configuration@Order(value = Ordered.HIGHEST_PRECEDENCE – 10)public class ThreadConfig { @Bean @Order(value = Ordered.HIGHEST_PRECEDENCE – 10) public ThreadPoolTaskExecutor testExecutor() { ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor(); threadPoolTaskExecutor.setThreadNamePrefix(“test-shutdown-pool-“); threadPoolTaskExecutor.setCorePoolSize(1); threadPoolTaskExecutor.setMaxPoolSize(1); threadPoolTaskExecutor.setKeepAliveSeconds(60); threadPoolTaskExecutor.setQueueCapacity(2000000); threadPoolTaskExecutor.setWaitForTasksToCompleteOnShutdown(true); threadPoolTaskExecutor.setAwaitTerminationSeconds(60); threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy()); return threadPoolTaskExecutor; }}

當然實際上的優(yōu)先級要根據(jù)情況調(diào)整，但是并沒有生效。后來看到一個說法，org.springframework.core.annotation.Order注解無法決定bean的創(chuàng)建順序，只能是bean創(chuàng)建完成后的一些業(yè)務的執(zhí)行時間。這個問題沒搞懂，反正Order未生效。那么只能抄他的代碼然后自己實現(xiàn)了。

該如何處理呢？

自實現(xiàn)

回到我們最初的代碼org.springframework.context.support.AbstractApplicationContext#doClose, 之前我們一直在看銷毀bean的邏輯，但是其實我們可以看到在此之前Spring發(fā)布了一個ContextClosedEvent事件，也就是說這個事件是早于Spring自己的bean銷毀邏輯前面的。

利用這個機制，我們可以將線程池的銷毀邏輯抄過來，并且監(jiān)聽ContextClosedEvent這個事件，然后在這個事件里執(zhí)行我們本地自己一些不被Spring管理的線程池的銷毀邏輯，正如前面看到的一樣。

一個簡單的例子如下

@Component@Slf4jpublic class ThreadPoolExecutorShutdownDefinition implements ApplicationListener { private static final List POOLS = Collections.synchronizedList(new ArrayList(12)); /** * 線程中的任務在接收到應用關(guān)閉信號量后最多等待多久就強制終止，其實就是給剩余任務預留的時間， 到時間后線程池必須銷毀 */ private static final long awaitTermination = 60; /** * awaitTermination的單位 */ private static final TimeUnit timeUnit = TimeUnit.SECONDS; /** * 注冊要關(guān)閉的線程池 * 注意如果調(diào)用這個方法的話，而線程池又是由Spring管理的，則必須等待這個bean初始化完成后才可以調(diào)用 * 因為依賴的{@link ThreadPoolTaskExecutor#getThreadPoolExecutor()}必須要在bean的父類方法中定義的 * 初始化{@link ExecutorConfigurationSupport#afterPropertiesSet()}方法中才會賦值 * * @param threadPoolTaskExecutor */ public static void registryExecutor(ThreadPoolTaskExecutor threadPoolTaskExecutor) { POOLS.add(threadPoolTaskExecutor.getThreadPoolExecutor()); } /** * 注冊要關(guān)閉的線程池 * 注意如果調(diào)用這個方法的話，而線程池又是由Spring管理的，則必須等待這個bean初始化完成后才可以調(diào)用 * 因為依賴的{@link ThreadPoolTaskExecutor#getThreadPoolExecutor()}必須要在bean的父類方法中定義的 * 初始化{@link ExecutorConfigurationSupport#afterPropertiesSet()}方法中才會賦值 * * 重寫了{@link ThreadPoolTaskScheduler#initializeExecutor(java.util.concurrent.ThreadFactory, java.util.concurrent.RejectedExecutionHandler)} * 來對父類的{@link ExecutorConfigurationSupport#executor}賦值 * * @param threadPoolTaskExecutor */ public static void registryExecutor(ThreadPoolTaskScheduler threadPoolTaskExecutor) { POOLS.add(threadPoolTaskExecutor.getScheduledThreadPoolExecutor()); } /** * 注冊要關(guān)閉的線程池， 如果一些線程池未交由線程池管理，則可以調(diào)用這個方法 * * @param executor */ public static void registryExecutor(ExecutorService executor) { POOLS.add(executor); } /** * 參考{@link org.springframework.scheduling.concurrent.ExecutorConfigurationSupport#shutdown()} * * @param event the event to respond to */ @Override public void onApplicationEvent(ContextClosedEvent event) { log.info(“容器關(guān)閉前處理線程池優(yōu)雅關(guān)閉開始, 當前要處理的線程池數(shù)量為: {} >>>>>>>>>>>>>>>>”, POOLS.size()); if (CollectionUtils.isEmpty(POOLS)) { return; } for (ExecutorService pool : POOLS) { pool.shutdown(); try { if (!pool.awaitTermination(awaitTermination, timeUnit)) { if (log.isWarnEnabled()) { log.warn(“Timed out while waiting for executor [{}] to terminate”, pool); } } } catch (InterruptedException ex) { if (log.isWarnEnabled()) { log.warn(“Timed out while waiting for executor [{}] to terminate”, pool); } Thread.currentThread().interrupt(); } } }}

如果想要本地的線程池實現(xiàn)優(yōu)雅停機，則直接調(diào)用上述對應的registryExecutor()方法即可，在容器銷毀的時候自然會去遍歷執(zhí)行對應邏輯。

做一點補充

我們所謂的優(yōu)雅停機， 必然是需要各方面的一些配合的。因為一個應用總歸最重要的還是外界的流量，上面只是處理了線程池的問題。

如果是普通的springboot項目， 停機無法解決流量繼續(xù)轉(zhuǎn)發(fā)進來的問題， 如nginx，只要保證發(fā)布時發(fā)送kill -15的信號量并且將發(fā)布機器從nginx負載中下線。

如果是Dubbo項目，則麻煩一些， 問題其實和上述類似，由于Dubbo也注冊了關(guān)閉的鉤子， 則在停機時會同時存在多個鉤子要執(zhí)行的問題。如果Spring的一些容器先銷毀了，Dubbo中的一些任務則無法繼續(xù)執(zhí)行。

java.lang.Runtime#addShutdownHook, 當存在多個注冊的關(guān)閉鉤子時，虛擬機會以某種未指定的順序并讓它們同時運行。這就是上述問題存在的原因。

public void addShutdownHook(Thread hook) { SecurityManager sm = System.getSecurityManager(); if (sm != null) { sm.checkPermission(new RuntimePermission(“shutdownHooks”)); } ApplicationShutdownHooks.add(hook);}

Dubbo應用則和上面手動監(jiān)聽容器銷毀事件的原理類似， 要讓Dubbo的鉤子先執(zhí)行,由于Dubbo已經(jīng)自己注冊了關(guān)閉鉤子，那么我們的步驟就變成了在Sprign容器啟動的時候先移除掉Dubbo的shutdownHook， 然后再容器銷毀的時候再添加回來。

綜合上面線程池的邏輯， 我們還要保證添加Dubbo的shutdownhook的Listener先執(zhí)行并且執(zhí)行完它的停機邏輯之后才執(zhí)行我們自己寫的處理線程池停機的Listener，當然如果線程池全部交由了Spring管理，自己沒有按照上面去重寫這一塊的邏輯， 則不需要注意這個問題。

移除和添加Dubbo的shutdownHook的邏輯類似如下.

public class DubboShutdownListener implements ApplicationListener, PriorityOrdered { private static final Logger LOGGER = LoggerFactory.getLogger(DubboShutdownListener.class); public DubboShutdownListener() { } public void onApplicationEvent(ApplicationEvent event) { if (event instanceof ApplicationStartedEvent) { Runtime.getRuntime().removeShutdownHook(DubboShutdownHook.getDubboShutdownHook()); LOGGER.info(“Dubbo default shutdown hook has been removed, It will be managed by Spring”); } else if (event instanceof ContextClosedEvent) { LOGGER.info(“Start destroy Dubbo Container on Spring close event”); DubboShutdownHook.getDubboShutdownHook().destroyAll(); LOGGER.info(“Dubbo Container has been destroyed finished”); } } public int getOrder() { return 0; }}