高并發之——建立線程池居然有這麼多方式...

前言

在Java的高并發領域，線程池一直是一個繞不開的話題。有些童鞋一直在使用線程池，但是，對于如何建立線程池僅僅停留在使用Executors工具類的方式，那麼，建立線程池究竟存在哪幾種方式呢？就讓我們一起從建立線程池的源碼來深入分析究竟有哪些方式可以建立線程池。

使用Executors工具類建立線程池

在建立線程池時，初學者用的最多的就是Executors 這個工具類，而使用這個工具類建立線程池時非常簡單的，不需要關注太多的線程池細節，隻需要傳入必要的參數即可。Executors 工具類提供了幾種建立線程池的方法，如下所示。

Executors.newCachedThreadPool：建立一個可緩存的線程池，如果線程池的大小超過了需要，可以靈活回收空閑線程，如果沒有可回收線程，則建立線程
Executors.newFixedThreadPool：建立一個定長的線程池，可以控制線程的最大并發數，超出的線程會在隊列中等待
Executors.newScheduledThreadPool：建立一個定長的線程池，支援定時、周期性的任務執行
Executors.newSingleThreadExecutor: 建立一個單線程化的線程池，使用一個唯一的工作線程執行任務，保證所有任務按照指定順序（先入先出或者優先級）執行
Executors.newSingleThreadScheduledExecutor:建立一個單線程化的線程池，支援定時、周期性的任務執行
Executors.newWorkStealingPool：建立一個具有并行級别的work-stealing線程池

其中，Executors.newWorkStealingPool方法是Java 8中新增的建立線程池的方法，它能夠為線程池設定并行級别，具有更高的并發度和性能。除了此方法外，其他建立線程池的方法本質上調用的是ThreadPoolExecutor類的構造方法。

例如，我們可以使用如下代碼建立線程池。

Executors.newWorkStealingPool();
Executors.newCachedThreadPool();
Executors.newScheduledThreadPool(3);

使用ThreadPoolExecutor類建立線程池

從代碼結構上看ThreadPoolExecutor類繼承自AbstractExecutorService，也就是說，ThreadPoolExecutor類具有AbstractExecutorService類的全部功能。

既然Executors工具類中建立線程池大部分調用的都是ThreadPoolExecutor類的構造方法，是以，我們也可以直接調用ThreadPoolExecutor類的構造方法來建立線程池，而不再使用Executors工具類。接下來，我們一起看下ThreadPoolExecutor類的構造方法。

ThreadPoolExecutor類中的所有構造方法如下所示。

public ThreadPoolExecutor(int corePoolSize,
                  int maximumPoolSize,
                  long keepAliveTime,
                  TimeUnit unit,
                 BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize,
                int maximumPoolSize,
                long keepAliveTime,
                TimeUnit unit,
                BlockingQueue<Runnable> workQueue,
                    ThreadFactory threadFactory) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
     threadFactory, defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize,
                int maximumPoolSize,
                long keepAliveTime,
                    TimeUnit unit,
                BlockingQueue<Runnable> workQueue,
                RejectedExecutionHandler handler) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
     Executors.defaultThreadFactory(), handler);
}
public ThreadPoolExecutor(int corePoolSize,
                int maximumPoolSize,
                long keepAliveTime,
                TimeUnit unit,
                    BlockingQueue<Runnable> workQueue,
                ThreadFactory threadFactory,
                RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.acc = System.getSecurityManager() == null ?
            null :
            AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

由ThreadPoolExecutor類的構造方法的源代碼可知，建立線程池最終調用的構造方法如下。

public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
              long keepAliveTime, TimeUnit unit,
              BlockingQueue<Runnable> workQueue,
              ThreadFactory threadFactory,
                  RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.acc = System.getSecurityManager() == null ?
            null :
            AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

關于此構造方法中各參數的含義和作用，各位可以移步《

高并發之——不得不說的線程池與ThreadPoolExecutor類淺析

》進行查閱。

大家可以自行調用ThreadPoolExecutor類的構造方法來建立線程池。例如，我們可以使用如下形式建立線程池。

new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                       60L, TimeUnit.SECONDS,
                       new SynchronousQueue<Runnable>());

使用ForkJoinPool類建立線程池

在Java8的Executors工具類中，新增了如下建立線程池的方式。

public static ExecutorService newWorkStealingPool(int parallelism) {
    return new ForkJoinPool
        (parallelism,
         ForkJoinPool.defaultForkJoinWorkerThreadFactory,
         null, true);
}
public static ExecutorService newWorkStealingPool() {
    return new ForkJoinPool
        (Runtime.getRuntime().availableProcessors(),
         ForkJoinPool.defaultForkJoinWorkerThreadFactory,
         null, true);
}

從源代碼可以可以，本質上調用的是ForkJoinPool類的構造方法類建立線程池，而從代碼結構上來看ForkJoinPool類繼承自AbstractExecutorService抽象類。接下來，我們看下ForkJoinPool類的構造方法。

public ForkJoinPool() {
    this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
         defaultForkJoinWorkerThreadFactory, null, false);
}
 public ForkJoinPool(int parallelism) {
    this(parallelism, defaultForkJoinWorkerThreadFactory, null, false);
}
public ForkJoinPool(int parallelism,
                ForkJoinWorkerThreadFactory factory,
                UncaughtExceptionHandler handler,
                boolean asyncMode) {
    this(checkParallelism(parallelism),
         checkFactory(factory),
         handler,
         asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
         "ForkJoinPool-" + nextPoolId() + "-worker-");
    checkPermission();
}
private ForkJoinPool(int parallelism,
                 ForkJoinWorkerThreadFactory factory,
                 UncaughtExceptionHandler handler,
                 int mode,
                 String workerNamePrefix) {
    this.workerNamePrefix = workerNamePrefix;
    this.factory = factory;
    this.ueh = handler;
    this.config = (parallelism & SMASK) | mode;
    long np = (long)(-parallelism); // offset ctl counts
    this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
}

通過檢視源代碼得知，ForkJoinPool的構造方法，最終調用的是如下私有構造方法。

private ForkJoinPool(int parallelism,
                 ForkJoinWorkerThreadFactory factory,
                 UncaughtExceptionHandler handler,
                 int mode,
                 String workerNamePrefix) {
    this.workerNamePrefix = workerNamePrefix;
    this.factory = factory;
    this.ueh = handler;
    this.config = (parallelism & SMASK) | mode;
    long np = (long)(-parallelism); // offset ctl counts
    this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
}

其中，各參數的含義如下所示。

parallelism：并發級别。
factory：建立線程的工廠類對象。
handler：當線程池中的線程抛出未捕獲的異常時，統一使用UncaughtExceptionHandler對象處理。
mode：取值為FIFO_QUEUE或者LIFO_QUEUE。
workerNamePrefix：執行任務的線程名稱的字首。

當然，私有構造方法雖然是參數最多的一個方法，但是其不會直接對外方法，我們可以使用如下方式建立線程池。

new ForkJoinPool();
new ForkJoinPool(Runtime.getRuntime().availableProcessors());
new ForkJoinPool(Runtime.getRuntime().availableProcessors(),
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
             null, true);

使用ScheduledThreadPoolExecutor類建立線程池

在Executors工具類中存在如下方法類建立線程池。

public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
    return new DelegatedScheduledExecutorService
        (new ScheduledThreadPoolExecutor(1));
}
public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) {
    return new DelegatedScheduledExecutorService
        (new ScheduledThreadPoolExecutor(1, threadFactory));
}
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}
public static ScheduledExecutorService newScheduledThreadPool(
        int corePoolSize, ThreadFactory threadFactory) {
    return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}

從源碼來看，這幾個方法本質上調用的都是ScheduledThreadPoolExecutor類的構造方法，ScheduledThreadPoolExecutor中存在的構造方法如下所示。

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}
public ScheduledThreadPoolExecutor(int corePoolSize, ThreadFactory threadFactory) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue(), threadFactory);
}
public ScheduledThreadPoolExecutor(int corePoolSize, RejectedExecutionHandler handler) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue(), handler);
}
public ScheduledThreadPoolExecutor(int corePoolSize,ThreadFactory threadFactory, RejectedExecutionHandler handler) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue(), threadFactory, handler);
}

而從代碼結構上看，ScheduledThreadPoolExecutor類繼承自ThreadPoolExecutor類，本質上還是調用ThreadPoolExecutor類的構造方法，隻不過此時傳遞的隊列為DelayedWorkQueue。我們可以直接調用ScheduledThreadPoolExecutor類的構造方法來建立線程池，例如以如下形式建立線程池。

new ScheduledThreadPoolExecutor(3)

最後，需要注意的是：ScheduledThreadPoolExecutor主要用來建立執行定時任務的線程池。

高并發之——建立線程池居然有這麼多方式...

前言

使用Executors工具類建立線程池

使用ThreadPoolExecutor類建立線程池

使用ForkJoinPool類建立線程池

使用ScheduledThreadPoolExecutor類建立線程池

繼續閱讀

關于Gradle配置的小結

Java小案例——随機數猜測随機數猜測

nginx location中斜線的位置的重要性

27 Best Free Eclipse Plug-ins for Java Developer to be ProductiveCode Quality PluginsText Editor PluginsDependency ManagementVersion Control Integration PluginsFramework Development Continuous Integration Related PluginsOther Utility Plugins

Java String.format方法的簡單使用

neo4j之cypher使用文檔

GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？

spark/scala關于【資源檔案】加載方法概述外部檔案加載方案測試資源檔案打包入jar包中小結

mybatis_入門程式Mybatis入門

AOP程式設計_Android優雅權限架構(1)概念基礎，2021金三銀四前言正文大綱正文

Effective Java 8:通用程式設計

OOM三種類型

工廠模式-三種類型

【遞歸】高效率求2的n次幂

win10本地scala和spark安裝安裝scala安裝spark

scala (3) Function 和 Method