ThreadPoolExecutor執行過程分析

ThreadPoolExecutor

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

• corePoolSize：線程池核心線程數（平時保留的線程數）
• maximumPoolSize：線程池最大線程數（當workQueue都放不下時，啓動新線程，最大線程數）
• keepAliveTime：超出corePoolSize數量的線程的保留時間。
• unit：keepAliveTime單位

• workQueue：阻塞隊列，存放來不及執行的線程

  • ArrayBlockingQueue：構造函數必定要傳大小
  • LinkedBlockingQueue：構造函數不傳大小會默認爲（Integer.MAX_VALUE ），當大量請求任務時，容易形成 內存耗盡。
  • SynchronousQueue：同步隊列，一個沒有存儲空間的阻塞隊列 ，將任務同步交付給工做線程。
  • PriorityBlockingQueue : 優先隊列
• threadFactory：線程工廠

• handler：飽和策略

  • AbortPolicy（默認）：直接拋棄
  • CallerRunsPolicy：用調用者的線程執行任務
  • DiscardOldestPolicy：拋棄隊列中最久的任務
  • DiscardPolicy：拋棄當前任務

ThreadPoolExecutor 測試代碼

//test.java
volatile int finishState = 0;


@Test
public void test4() throws InterruptedException, ExecutionException {
   ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(3, 7, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<>(10));
   ExecutorCompletionService<String> executorCompletionService = new ExecutorCompletionService(threadPoolExecutor);

   Runnable runnable = new Runnable() {
      @Override
      public void run() {
         for (int i = 0; i < 50; i++) {
            String name = "name_" + i;
            TestCallable testCallable = new TestCallable(name);
            try {
               executorCompletionService.submit(testCallable);

               synchronized (lock) {
                  System.out.print("+++添加任務 name: " + name);
                  System.out.print(" ActiveCount: " + threadPoolExecutor.getActiveCount());
                  System.out.print(" poolSize: " + threadPoolExecutor.getPoolSize());
                  System.out.print(" queueSize: " + threadPoolExecutor.getQueue().size());
                  System.out.println(" taskCount: " + threadPoolExecutor.getTaskCount());
               }
            } catch (RejectedExecutionException e) {
               synchronized (lock) {
                  System.out.println("拒絕：" + name);
               }
            }
            try {
               Thread.sleep(200);
            } catch (InterruptedException e) {
               e.printStackTrace();
            }
         }
         finishState = 1;
      }
   };

   Thread addThread = new Thread(runnable);
   addThread.start();

   //System.out.println(" taskCount: " + threadPoolExecutor.getTaskCount());

   //添加的任務有被拋棄的。taskCount不必定等於添加的任務。
   int completeCount = 0;
   while (!(completeCount == threadPoolExecutor.getTaskCount() && finishState == 1)) {
      Future<String> take = executorCompletionService.take();
      String taskName = take.get();
      synchronized (lock) {
         System.out.print("---完成任務 name: " + taskName);
         System.out.print(" ActiveCount: " + threadPoolExecutor.getActiveCount());
         System.out.print(" poolSize: " + threadPoolExecutor.getPoolSize());
         System.out.print(" queueSize: " + threadPoolExecutor.getQueue().size());
         System.out.print(" taskCount: " + threadPoolExecutor.getTaskCount());
         System.out.println(" finishTask：" + (++completeCount));

      }
   }

   addThread.join();


   while (threadPoolExecutor.getPoolSize() > 0) {
      Thread.sleep(1000);
      synchronized (lock) {
         SimpleDateFormat simpleDateFormat = new SimpleDateFormat("HH:mm:ss");
         System.out.print(simpleDateFormat.format(new Date()));
         //System.out.print("name: " + taskName);
         System.out.print(" ActiveCount: " + threadPoolExecutor.getActiveCount());
         System.out.print(" poolSize: " + threadPoolExecutor.getPoolSize());
         System.out.print(" queueSize: " + threadPoolExecutor.getQueue().size());
         System.out.println(" taskCount: " + threadPoolExecutor.getTaskCount());
      }
   }

   // Tell threads to finish off.
   threadPoolExecutor.shutdown();
   // Wait for everything to finish.
   while (!threadPoolExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
      System.out.println("complete");
   }

}

//TestCallable.java
public class TestCallable implements Callable<String>{


   private String name;

   public TestCallable(String name) {
      this.name = name;
   }

   @Override
   public String call() throws Exception {
      try {
         Thread.sleep(2000);
      } catch (InterruptedException e) {
         e.printStackTrace();
      }
      return this.name;
   }
}

執行結果

+++添加任務 name: name_0 ActiveCount: 1 poolSize: 1 queueSize: 0 taskCount: 1
+++添加任務 name: name_1 ActiveCount: 2 poolSize: 2 queueSize: 0 taskCount: 2
+++添加任務 name: name_2 ActiveCount: 3 poolSize: 3 queueSize: 0 taskCount: 3
//corePoolSize=3，因此poolSize逐漸加到3.
+++添加任務 name: name_3 ActiveCount: 3 poolSize: 3 queueSize: 1 taskCount: 4
+++添加任務 name: name_4 ActiveCount: 3 poolSize: 3 queueSize: 2 taskCount: 5
+++添加任務 name: name_5 ActiveCount: 3 poolSize: 3 queueSize: 3 taskCount: 6
+++添加任務 name: name_6 ActiveCount: 3 poolSize: 3 queueSize: 4 taskCount: 7
+++添加任務 name: name_7 ActiveCount: 3 poolSize: 3 queueSize: 5 taskCount: 8
+++添加任務 name: name_8 ActiveCount: 3 poolSize: 3 queueSize: 6 taskCount: 9
+++添加任務 name: name_9 ActiveCount: 3 poolSize: 3 queueSize: 7 taskCount: 10
---完成任務 name: name_0 ActiveCount: 3 poolSize: 3 queueSize: 6 taskCount: 10 finishTask：1
+++添加任務 name: name_10 ActiveCount: 3 poolSize: 3 queueSize: 7 taskCount: 11
---完成任務 name: name_1 ActiveCount: 3 poolSize: 3 queueSize: 6 taskCount: 11 finishTask：2
+++添加任務 name: name_11 ActiveCount: 3 poolSize: 3 queueSize: 7 taskCount: 12
---完成任務 name: name_2 ActiveCount: 3 poolSize: 3 queueSize: 6 taskCount: 12 finishTask：3
+++添加任務 name: name_12 ActiveCount: 3 poolSize: 3 queueSize: 7 taskCount: 13
+++添加任務 name: name_13 ActiveCount: 3 poolSize: 3 queueSize: 8 taskCount: 14
+++添加任務 name: name_14 ActiveCount: 3 poolSize: 3 queueSize: 9 taskCount: 15
+++添加任務 name: name_15 ActiveCount: 3 poolSize: 3 queueSize: 10 taskCount: 16
//由於任務隊列爲：new LinkedBlockingDeque<>(10)，core線程不夠用時，將任務添加到隊列中。
//queueSize到10，隊列已滿。開始增長poolSize，加到maximumPoolSize=7
+++添加任務 name: name_16 ActiveCount: 4 poolSize: 4 queueSize: 10 taskCount: 17
+++添加任務 name: name_17 ActiveCount: 5 poolSize: 5 queueSize: 10 taskCount: 18
+++添加任務 name: name_18 ActiveCount: 6 poolSize: 6 queueSize: 10 taskCount: 19
+++添加任務 name: name_19 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 20
---完成任務 name: name_3 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 20 finishTask：4
+++添加任務 name: name_20 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 21
---完成任務 name: name_4 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 21 finishTask：5
+++添加任務 name: name_21 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 22
---完成任務 name: name_5 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 22 finishTask：6
+++添加任務 name: name_22 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 23
//poolSize=maximumPoolSize,queueSize=10，都滿了。執行飽和策略。默認AbortPolicy拋棄新任務。
拒絕：name_23
拒絕：name_24
拒絕：name_25
---完成任務 name: name_16 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 23 finishTask：7
+++添加任務 name: name_26 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 24
---完成任務 name: name_17 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 24 finishTask：8
+++添加任務 name: name_27 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 25
---完成任務 name: name_18 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 25 finishTask：9
+++添加任務 name: name_28 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 26
---完成任務 name: name_19 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 26 finishTask：10
+++添加任務 name: name_29 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 27
---完成任務 name: name_6 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 27 finishTask：11
+++添加任務 name: name_30 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 28
---完成任務 name: name_7 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 28 finishTask：12
+++添加任務 name: name_31 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 29
---完成任務 name: name_8 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 29 finishTask：13
+++添加任務 name: name_32 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 30
拒絕：name_33
拒絕：name_34
拒絕：name_35
---完成任務 name: name_9 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 30 finishTask：14
+++添加任務 name: name_36 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 31
---完成任務 name: name_10 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 31 finishTask：15
+++添加任務 name: name_37 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 32
---完成任務 name: name_11 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 32 finishTask：16
+++添加任務 name: name_38 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 33
---完成任務 name: name_12 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 33 finishTask：17
+++添加任務 name: name_39 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 34
---完成任務 name: name_13 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 34 finishTask：18
+++添加任務 name: name_40 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 35
---完成任務 name: name_14 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 35 finishTask：19
+++添加任務 name: name_41 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 36
---完成任務 name: name_15 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 36 finishTask：20
+++添加任務 name: name_42 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 37
拒絕：name_43
拒絕：name_44
拒絕：name_45
---完成任務 name: name_20 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 37 finishTask：21
+++添加任務 name: name_46 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 38
---完成任務 name: name_21 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 38 finishTask：22
+++添加任務 name: name_47 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 39
---完成任務 name: name_22 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 39 finishTask：23
//消費速度大於添加速度，poolSize並不減小，先把queue中的任務逐一執行完成。
+++添加任務 name: name_48 ActiveCount: 7 poolSize: 7 queueSize: 10 taskCount: 40
---完成任務 name: name_26 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 40 finishTask：24
---完成任務 name: name_27 ActiveCount: 7 poolSize: 7 queueSize: 8 taskCount: 40 finishTask：25
+++添加任務 name: name_49 ActiveCount: 7 poolSize: 7 queueSize: 9 taskCount: 41
---完成任務 name: name_28 ActiveCount: 7 poolSize: 7 queueSize: 8 taskCount: 41 finishTask：26
---完成任務 name: name_29 ActiveCount: 7 poolSize: 7 queueSize: 7 taskCount: 41 finishTask：27
---完成任務 name: name_30 ActiveCount: 7 poolSize: 7 queueSize: 6 taskCount: 41 finishTask：28
---完成任務 name: name_31 ActiveCount: 7 poolSize: 7 queueSize: 5 taskCount: 41 finishTask：29
---完成任務 name: name_32 ActiveCount: 7 poolSize: 7 queueSize: 4 taskCount: 41 finishTask：30
---完成任務 name: name_36 ActiveCount: 7 poolSize: 7 queueSize: 3 taskCount: 41 finishTask：31
---完成任務 name: name_37 ActiveCount: 7 poolSize: 7 queueSize: 2 taskCount: 41 finishTask：32
---完成任務 name: name_38 ActiveCount: 7 poolSize: 7 queueSize: 1 taskCount: 41 finishTask：33
---完成任務 name: name_39 ActiveCount: 7 poolSize: 7 queueSize: 0 taskCount: 41 finishTask：34
//ActiveCount開始減小。
---完成任務 name: name_40 ActiveCount: 6 poolSize: 7 queueSize: 0 taskCount: 41 finishTask：35
---完成任務 name: name_41 ActiveCount: 5 poolSize: 7 queueSize: 0 taskCount: 41 finishTask：36
---完成任務 name: name_42 ActiveCount: 4 poolSize: 7 queueSize: 0 taskCount: 41 finishTask：37
---完成任務 name: name_46 ActiveCount: 3 poolSize: 7 queueSize: 0 taskCount: 41 finishTask：38
---完成任務 name: name_47 ActiveCount: 2 poolSize: 7 queueSize: 0 taskCount: 41 finishTask：39
---完成任務 name: name_48 ActiveCount: 1 poolSize: 7 queueSize: 0 taskCount: 41 finishTask：40
---完成任務 name: name_49 ActiveCount: 0 poolSize: 7 queueSize: 0 taskCount: 41 finishTask：41
22:47:49 ActiveCount: 0 poolSize: 7 queueSize: 0 taskCount: 41
22:47:50 ActiveCount: 0 poolSize: 7 queueSize: 0 taskCount: 41
22:47:51 ActiveCount: 0 poolSize: 7 queueSize: 0 taskCount: 41
22:47:52 ActiveCount: 0 poolSize: 7 queueSize: 0 taskCount: 41
22:47:53 ActiveCount: 0 poolSize: 7 queueSize: 0 taskCount: 41
22:47:54 ActiveCount: 0 poolSize: 7 queueSize: 0 taskCount: 41
22:47:55 ActiveCount: 0 poolSize: 7 queueSize: 0 taskCount: 41
22:47:56 ActiveCount: 0 poolSize: 7 queueSize: 0 taskCount: 41
22:47:57 ActiveCount: 0 poolSize: 5 queueSize: 0 taskCount: 41
//因爲keepAliveTime=10，大概10s後poolSize恢復爲corePoolSize。
22:47:58 ActiveCount: 0 poolSize: 3 queueSize: 0 taskCount: 41
22:47:59 ActiveCount: 0 poolSize: 3 queueSize: 0 taskCount: 41
22:48:00 ActiveCount: 0 poolSize: 3 queueSize: 0 taskCount: 41

注意

《阿里巴巴java開發手冊》

線程池不使用 Executors 去建立，而是經過 ThreadPoolExecutor 的方式，這樣 的處理方式讓寫的同窗更加明確線程池的運行規則，規避資源耗盡的風險。

說明： Executors 返回的線程池對象的弊端以下：

1. FixedThreadPool 和 SingleThreadPool : 容許的請求隊列長度爲 Integer.MAX_VALUE ，可能會堆積大量的請求，從而致使 OOM 。
2. CachedThreadPool 和 ScheduledThreadPool : 容許的建立線程數量爲 Integer.MAX_VALUE ，可能會建立大量的線程，從而致使 OOM 。

Reference

• 深刻淺出Java線程池