Executor - Shutdown、ShutdownNow、awaitTermination 详解与实战

/**
     * Initiates an orderly shutdown in which previously submitted
     * tasks are executed, but no new tasks will be accepted.
     * Invocation has no additional effect if already shut down.
     *
     * <p>This method does not wait for previously submitted tasks to
     * complete execution.  Use {@link #awaitTermination awaitTermination}
     * to do that.
     *
     * @throws SecurityException {@inheritDoc}
     */
    public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            advanceRunState(SHUTDOWN);
            interruptIdleWorkers();
            onShutdown(); // hook for ScheduledThreadPoolExecutor
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
    }

/**
     * Attempts to stop all actively executing tasks, halts the
     * processing of waiting tasks, and returns a list of the tasks
     * that were awaiting execution. These tasks are drained (removed)
     * from the task queue upon return from this method.
     *
     * <p>This method does not wait for actively executing tasks to
     * terminate.  Use {@link #awaitTermination awaitTermination} to
     * do that.
     *
     * <p>There are no guarantees beyond best-effort attempts to stop
     * processing actively executing tasks.  This implementation
     * cancels tasks via {@link Thread#interrupt}, so any task that
     * fails to respond to interrupts may never terminate.
     *
     * @throws SecurityException {@inheritDoc}
     */
    public List<Runnable> shutdownNow() {
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            advanceRunState(STOP);
            interruptWorkers();
            tasks = drainQueue();
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
        return tasks;
    }

* Threads waiting in awaitTermination() will return when the
     * state reaches TERMINATED.
    public boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            for (;;) {
                if (runStateAtLeast(ctl.get(), TERMINATED))
                    return true;
                if (nanos <= 0)
                    return false;
                nanos = termination.awaitNanos(nanos);
            }
        } finally {
            mainLock.unlock();
        }
    }

import java.util.concurrent.{CopyOnWriteArraySet, LinkedBlockingQueue, ThreadPoolExecutor, TimeUnit}
object ExecutorPoolShutdown {
  // Runnable 内执行逻辑，寻找上下界
  def processNumBuffer(nums: Array[Int], taskId: Long): String = {
    val maxNum = nums.max
    val minNum = nums.min
    val log = s"TaskId: $taskId Length: ${nums.length} Min: $minNum Max: $maxNum"
    log
  }
  def main(args: Array[String]): Unit = {
    // 存储所有 Task 的日志
    val logAll = new StringBuffer()
    // 存储所有可用的 TaskId
    val taskSet = new CopyOnWriteArraySet[Long]()
    // 初始化线程池
    val executor = new ThreadPoolExecutor(6, 10, 3000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue())
    val numIterator = (0 until 500000).iterator
    // 每50000个数据生成一个 Task
    numIterator.grouped(50000).foreach(group => {
      executor.execute(new Runnable() {
        override def run(): Unit = {
          val taskId = Thread.currentThread().getId
          taskSet.add(taskId) // 添加 taskID
          val res = processNumBuffer(group.toArray, taskId)
          logAll.append(res + "\n") // 添加统计日志
        }
      })
    })
    // 调用 shutdown
    executor.shutdown()
    // 获取当前线程
    println("After shutdown...")
    getCurrentThread()
    val st = System.currentTimeMillis()
    // 调用 awaitTerminatio
    val state = executor.awaitTermination(500, TimeUnit.MILLISECONDS)
    val cost = System.currentTimeMillis() - st
    println(s"Executor State: $state Cost: $cost")
    // 再次获取当前线程
    println("After awaitTermination...")
    getCurrentThread()
    println(logAll.toString)
    println(taskSet.toArray().mkString(","))
  }
}

numIterator.grouped(50000).foreach(group => {
      executor.execute(new Runnable() {
        override def run(): Unit = {
          val taskId = Thread.currentThread().getId
          taskSet.add(taskId)
          Thread.sleep(2000)
          val res = processNumBuffer(group.toArray, taskId)
          logAll.append(res + "\n")
        }
      })
    })

val st = System.currentTimeMillis()
    val state = executor.awaitTermination(2000, TimeUnit.MILLISECONDS)
    val cost = System.currentTimeMillis() - st
    println(s"Executor State: $state Cost: $cost")

val tasks = executor.shutdownNow()
    tasks.asScala.foreach(task => {
      println(task)
    })

numIterator.grouped(50000).foreach(group => {
      executor.execute(new Runnable() {
        override def run(): Unit = {
          val taskId = Thread.currentThread().getId
          taskSet.add(taskId)
          Thread.sleep(2000)
          val res = processNumBuffer(group.toArray, taskId)
          logAll.append(res + "\n")
        }
      })
    })

Caused by: java.lang.InterruptedException: sleep interrupted

val st = System.currentTimeMillis()
    val state = executor.awaitTermination(3000, TimeUnit.MILLISECONDS)
    val cost = System.currentTimeMillis() - st
    println(s"Executor State: $state Cost: $cost")
    println("After awaitTermination...")
    getCurrentThread()
    val tasks = executor.shutdownNow()
    tasks.asScala.foreach(task => {
      println(task)
    })
    println("After shutdown...")
    getCurrentThread()

executor.shutdown()
      while (!executor.awaitTermination(500, TimeUnit.MILLISECONDS)) {
         println("Task is Running...")
      }
      println("Task is Finish!")