Java 執行緒池詳解及建立簡單實例

Java 執行緒池

最近在改進專案的並發功能，但開發起來磕磕碰碰的。看好多資料，總算加深了認識。於是打算配合查看原始碼，總結並發程式設計的原理。

準備從用得最多的執行緒池開始，圍繞著創建、執行、關閉認識執行緒池整個生命週期的實作原理。後續再研究原子變數、並發容器、阻塞佇列、同步工具、鎖等主題。 java.util.concurrent裡的並發工具用起來不難，但不能只是會用，我們要read the fucking source code，哈哈。順便說聲，我用的JDK是1.8。

Executor框架

Executor是一套執行緒池管理框架，介面裡只有一個方法execute，執行Runnable任務。 ExecutorService介面擴展了Executor，增加了執行緒生命週期的管理，提供任務終止、傳回任務結果等方法。 AbstractExecutorService實作了ExecutorService，提供例如submit方法的預設實作邏輯。

然後到今天的主題ThreadPoolExecutor，繼承了AbstractExecutorService，提供線程池的具體實作。

建構法

下面是ThreadPoolExecutor最普通的建構函數，最多有七個參數。具體程式碼不貼了，只是一些參數校驗設定的語句。

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

corePoolSize是執行緒池的目標大小，也就是執行緒池剛剛建立起來，還沒有任務要執行時的大小。 maximumPoolSize是執行緒池的最大上限。 keepAliveTime是執行緒的存活時間，當執行緒池內的執行緒數大於corePoolSize，超出存活時間的空閒執行緒就會被回收。 unit就不用說了，剩下的三個參數看後文的分析。

預設的定制線程池

ThreadPoolExecutor預設了一些已經定制好的線程池，由Executors裡的工廠方法創建。以下分析newSingleThreadExecutor、newFixedThreadPool、newCachedThreadPool的建立參數。

newFixedThreadPool

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                   0L, TimeUnit.MILLISECONDS,
                   new LinkedBlockingQueue<Runnable>());
  }

newFixedThreadPool的corePoolSize和maximumPoolSize都設定為傳入的固定數量，keepAliveTim設定為0。執行緒池建立後，執行緒數量將會固定不變，適合需要執行緒很穩定的場合。

newSingleThreadExecutor

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
      (new ThreadPoolExecutor(1, 1,
                  0L, TimeUnit.MILLISECONDS,
                  new LinkedBlockingQueue<Runnable>()));
  }

newSingleThreadExecutor是執行緒數量固定為1的newFixedThreadPool版本，保證池內的任務序列。注意到返回的是FinalizableDelegatedExecutorService，來看看源碼：

static class FinalizableDelegatedExecutorService
    extends DelegatedExecutorService {
    FinalizableDelegatedExecutorService(ExecutorService executor) {
      super(executor);
    }
    protected void finalize() {
      super.shutdown();
    }
  }

FinalizableDelegatedExecutorService繼承了DelegatedExecutorService，僅僅在gc時增加關閉線程池的操作，再來看看DelegatedExecutorService的源碼：簡單的包裝：De錶達使其只暴露出ExecutorService的方法，因此不能再配置線程池的參數。本來，線程池創建的參數是可以調整的，ThreadPoolExecutor提供了set方法。使用newSingleThreadExecutor目的是產生單執行緒串列的執行緒池，如果還能配置執行緒池大小，那就沒意思了。

Executors也提供了unconfigurableExecutorService方法，將普通執行緒池包裝成不可設定的執行緒池。如果不想線程池被不明所以的後人修改，可以呼叫這個方法。

newCachedThreadPool

static class DelegatedExecutorService extends AbstractExecutorService {
    private final ExecutorService e;
    DelegatedExecutorService(ExecutorService executor) { e = executor; }
    public void execute(Runnable command) { e.execute(command); }
    public void shutdown() { e.shutdown(); }
    public List<Runnable> shutdownNow() { return e.shutdownNow(); }
    public boolean isShutdown() { return e.isShutdown(); }
    public boolean isTerminated() { return e.isTerminated(); }
    //...
  }

newCachedThreadPool產生一個會快取的執行緒池，執行緒數量可以從0到Integer.MAX_VALUE，逾時時間為1分鐘。線程池用起來的效果是：如果有空閒線程，會重複使用線程；如果沒有空閒線程，會新線程；如果線程空閒超過1分鐘，將會被回收。

newScheduledThreadPool

newScheduledThreadPool將會建立一個可定時執行任務的執行緒池。這個不打算在本文展開，後續會另開文章細講。

等待佇列

newCachedThreadPool的執行緒上限幾乎等同於無限，但係統資源是有限的，任務的處理速度總有可能比不上任務的提交速度。因此，可以為ThreadPoolExecutor提供一個阻塞佇列來保存因線程不足而等待的Runnable任務，這就是BlockingQueue。

JDK為BlockingQueue提供了幾種實作方式，常用的有：

ArrayBlockingQueue：陣列結構的阻塞佇列

LinkedBlockingQueue：鍊錶結構的阻塞佇列

ynue儲存元素的阻塞佇列

newFixedThreadPool和newSingleThreadExecutor在預設情況下使用一個無界的LinkedBlockingQueue。要注意的是，如果任務一直提交，但執行緒池又不能及時處理，等待佇列將會無限制地加長，系統資源總會有消耗殆盡的一刻。所以，建議使用有界的等待隊列，避免資源耗盡。但解決一個問題，又會帶來新問題：隊列填滿之後，再來新任務，這個時候該怎麼辦？後文會介紹如何處理隊列飽和。

newCachedThreadPool使用的SynchronousQueue十分有趣，看名称是个队列，但它却不能存储元素。要将一个任务放进队列，必须有另一个线程去接收这个任务，一个进就有一个出，队列不会存储任何东西。因此，SynchronousQueue是一种移交机制，不能算是队列。newCachedThreadPool生成的是一个没有上限的线程池，理论上提交多少任务都可以，使用SynchronousQueue作为等待队列正合适。

饱和策略

当有界的等待队列满了之后，就需要用到饱和策略去处理，ThreadPoolExecutor的饱和策略通过传入RejectedExecutionHandler来实现。如果没有为构造函数传入，将会使用默认的defaultHandler。

private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
public static class AbortPolicy implements RejectedExecutionHandler {
    public AbortPolicy() { }
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
      throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + e.toString());
    }
  }

AbortPolicy是默认的实现，直接抛出一个RejectedExecutionException异常，让调用者自己处理。除此之外，还有几种饱和策略，来看一下：

public static class DiscardPolicy implements RejectedExecutionHandler {
   public DiscardPolicy() { }
   public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
   }
 }

DiscardPolicy的rejectedExecution直接是空方法，什么也不干。如果队列满了，后续的任务都抛弃掉。

public static class DiscardOldestPolicy implements RejectedExecutionHandler {
   public DiscardOldestPolicy() { }
   public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
     if (!e.isShutdown()) {
       e.getQueue().poll();
       e.execute(r);
     }
   }
 }

DiscardOldestPolicy会将等待队列里最旧的任务踢走，让新任务得以执行。

public static class CallerRunsPolicy implements RejectedExecutionHandler {
   public CallerRunsPolicy() { }
   public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
     if (!e.isShutdown()) {
       r.run();
     }
   }
 }

最后一种饱和策略是CallerRunsPolicy，它既不抛弃新任务，也不抛弃旧任务，而是直接在当前线程运行这个任务。当前线程一般就是主线程啊，让主线程运行任务，说不定就阻塞了。如果不是想清楚了整套方案，还是少用这种策略为妙。

ThreadFactory

每当线程池需要创建一个新线程，都是通过线程工厂获取。如果不为ThreadPoolExecutor设定一个线程工厂，就会使用默认的defaultThreadFactory：

public static ThreadFactory defaultThreadFactory() {
  return new DefaultThreadFactory();
}

static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;
 
    DefaultThreadFactory() {
      SecurityManager s = System.getSecurityManager();
      group = (s != null) ? s.getThreadGroup() :
                 Thread.currentThread().getThreadGroup();
      namePrefix = "pool-" +
             poolNumber.getAndIncrement() +
            "-thread-";
    }
 
    public Thread newThread(Runnable r) {
      Thread t = new Thread(group, r,
                 namePrefix + threadNumber.getAndIncrement(),
                 0);
      if (t.isDaemon())
        t.setDaemon(false);
      if (t.getPriority() != Thread.NORM_PRIORITY)
        t.setPriority(Thread.NORM_PRIORITY);
      return t;
    }
  }

   

平时打印线程池里线程的name时，会输出形如pool-1-thread-1之类的名称，就是在这里设置的。这个默认的线程工厂，创建的线程是普通的非守护线程，如果需要定制，实现ThreadFactory后传给ThreadPoolExecutor即可。

不看代码不总结不会知道，光是线程池的创建就可以引出很多学问。别看平时创建线程池是一句代码的事，其实ThreadPoolExecutor提供了很灵活的定制方法。

感谢阅读，希望能帮助到大家，谢谢大家对本站的支持！

更多Java 线程池详解及创建简单实例相关文章请关注PHP中文网！