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

探究

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

我們知道Spring已經實現了自己的優雅停機方案，詳細請參考org.springframework.context.support.AbstractApplicationContext#registerShutdownHook，然后主要看調用的org.springframework.context.support.AbstractApplicationContext#doClose，在這個方法里定義了容器銷毀的執行順序

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

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

 private final Map<String, Object> 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();
 }

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

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

相關代碼貼一下

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

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

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

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

@Configuration
public 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;
    }
}

現在來一下線程池的這個類結構，ThreadPoolTaskExecutor繼承了org.springframework.scheduling.concurrent.ExecutorConfigurationSupport, 實現了org.springframework.beans.factory.DisposableBean,完整結構如下

public class ThreadPoolTaskExecutor extends ExecutorConfigurationSupport
  implements AsyncListenableTaskExecutor, SchedulingTaskExecutor {
  
}

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

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

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

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

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

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

畢竟我們在創建bean的時候是直接new的ThreadPoolTaskExecutor，并沒有去處理這個屬性。還是看線程池的父類ExecutorConfigurationSupport, 其實現了接口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在初始化完成后會調用父類的afterPropertiesSet方法，上面的代碼已經很清晰的說明了問題，最終父類中又定義了抽象方法initializeExecutor()，供子類去具體實現如果初始化這個屬性executor，因為我們知道線程池的實現除了普通的異步任務線程池ThreadPoolTaskExecutor, 還有基于定時調度的線程池ThreadPoolTaskExecutor，具體實現這里就不貼出來了，反正已經能夠看出來這個屬性是如何被賦值的了，所以上述銷毀時代碼可以直接使用。

現在整體總結下來，其實發現我們Spring已經幫我們實現了線程池的優雅停機規則，在接收到停機信號時，先拒絕接收新的任務，并繼續執行已經接受的任務，在任務執行完成或者到達最大等待時間，完成線程池的關閉。這么一整套邏輯正是我們所需要的，而我們如果要使用這個邏輯，僅僅需要在配置線程池的時候指定下上面看到的waitForTasksToCompleteOnShutdown屬性和awaitTerminationMillis屬性。

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

@Configuration
public 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;
    }
}

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

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

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

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

之前沒有貼出來具體代碼，現在來看下org.springframework.beans.factory.support.AbstractBeanFactory#requiresDestruction方法中的調用的另外一個本地方法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);
 }

之前線程池能夠執行銷毀流程是因為實現了接口DisposableBean，而數據源則是實現了另外一個接口AutoCloseable，因此數據源也是會執行銷毀邏輯的。

現在我們只要介入bean的創建優先級即可，使用org.springframework.core.annotation.Order注解來指定線程池創建的高優先級，如下。

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

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

該如何處理呢？

自實現

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

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

一個簡單的例子如下

@Component
@Slf4j
public class ThreadPoolExecutorShutdownDefinition implements ApplicationListener<ContextClosedEvent> {

    private static final List<ExecutorService> POOLS = Collections.synchronizedList(new ArrayList<>(12));

    /**
     * 線程中的任務在接收到應用關閉信號量后最多等待多久就強制終止，其實就是給剩余任務預留的時間， 到時間后線程池必須銷毀
     */
    private static final long awaitTermination = 60;

    /**
     * awaitTermination的單位
     */
    private static final TimeUnit timeUnit = TimeUnit.SECONDS;


    /**
     * 注冊要關閉的線程池
     * 注意如果調用這個方法的話，而線程池又是由Spring管理的，則必須等待這個bean初始化完成后才可以調用
     * 因為依賴的{@link ThreadPoolTaskExecutor#getThreadPoolExecutor()}必須要在bean的父類方法中定義的
     * 初始化{@link ExecutorConfigurationSupport#afterPropertiesSet()}方法中才會賦值
     *
     * @param threadPoolTaskExecutor
     */
    public static void registryExecutor(ThreadPoolTaskExecutor threadPoolTaskExecutor) {
        POOLS.add(threadPoolTaskExecutor.getThreadPoolExecutor());
    }

    /**
     * 注冊要關閉的線程池
     * 注意如果調用這個方法的話，而線程池又是由Spring管理的，則必須等待這個bean初始化完成后才可以調用
     * 因為依賴的{@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());
    }

    /**
     * 注冊要關閉的線程池， 如果一些線程池未交由線程池管理，則可以調用這個方法
     *
     * @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("容器關閉前處理線程池優雅關閉開始, 當前要處理的線程池數量為: {} >>>>>>>>>>>>>>>>", 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();
            }
        }
    }
}

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

做一點補充

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

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

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

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

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

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

綜合上面線程池的邏輯，我們還要保證添加Dubbo的shutdownhook的Listener先執行并且執行完它的停機邏輯之后才執行我們自己寫的處理線程池停機的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;
    }
}

日日操夜夜添-日日操影院-日日草夜夜操-日日干干-精品一区二区三区波多野结衣-精品一区二区三区高清免费不卡

如何優雅的關閉 Java線程池

探究

自實現

做一點補充

數獨大挑戰2018-06-03

答題星2018-06-03

全階人生考試2018-06-03

運動步數有氧達人2018-06-03

每日養生app2018-06-03

體育訓練成績評定2018-06-03