Java并发编程之线程池的运用

2019-11-18杂谈搜奇网48°c

A⁺ A^-

1. 为何要运用多线程？

跟着科技的进步，如今的电脑及服务器的处置惩罚器数目都比较多，今后可能会越来越多，比方我的事情电脑的处置惩罚器有8个，怎样检察呢？

计算机右键--属性--装备治理器，翻开属性窗口，然后点击“装备治理器”，在“处置惩罚器”下可看到一切的处置惩罚器：

也能够经由过程以下Java代码猎取到处置惩罚器的个数：

System.out.println("CPU个数:" + Runtime.getRuntime().availableProcessors());

运转效果以下所示：

CPU个数:8

既然处置惩罚器的个数增添了，假如还运用传统的串行编程，就有点浪费资本了，因而，为了进步资本利用率，让各个处置惩罚器都劳碌起来，就须要引入并发编程，要引入并发编程，就引入了多线程。

能够说，运用多线程的最直接目标就是为了进步资本利用率，资本的利用率进步了，体系的吞吐率也就相应进步了。

2. 为何要运用线程池？

在肯定的局限内，增添线程能够进步应用递次的吞吐率，但线程并非越多越好（因为线程的建立与烧毁都须要很大的开支），假如凌驾了某个局限，不仅会下降应用递次的实行速率，严峻的话，应用递次甚至会崩溃，以至于不能不重启应用递次。

为了防备这类题目，就须要对应用递次能够建立的线程数目举行限定，确保在线程数目抵达限定时，递次也不会耗尽资本，线程池就是为了处置惩罚这类题目而涌现的。

线程池：治理一组事情线程的资本池。

线程池与事情行列密切相关，事情行列中保留了一切守候实行的使命。

事情者线程的使命就是从事情行列中猎取一个使命，实行使命，然后返回线程池并守候下一个使命。

运用线程池能够带来以下优点：

经由过程重用现有的线程而不是建立新线程，能够在处置惩罚多个使命时削减在线程建立与烧毁过程当中发生的庞大开支。
当使命抵达时，事情线程一般已存在，因而不会因为守候建立线程而耽误使命的实行，从而进步了相应性。
能够经由过程调解线程池的大小，建立足够多的线程使处置惩罚器坚持劳碌状况，同时还能够防备过量线程相互竞争资本而使应用递次耗尽内存或崩溃。

3. 建立线程池

3.1 运用Executors静态要领建立（不引荐）

Executors类供应了以下4个静态要领来疾速的建立线程池：

newFixedThreadPool
newCachedThreadPool
newSingleThreadExecutor
newScheduledThreadPool

起首看下newFixedThreadPool()要领的运用体式格局：

ExecutorService threadPool = Executors.newFixedThreadPool(10);

它的源码为：

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

申明：newFixedThreadPool将建立一个牢固长度的线程池，每当提交一个使命时就建立一个线程，直到抵达线程池的最大数目，这时候线程池的范围将不再变化（假如某个线程因为发生了未预期的Exception而终了，那末线程池会补充一个新的线程）。

然后看下newCachedThreadPool()要领的运用体式格局：

ExecutorService threadPool = Executors.newCachedThreadPool();

它的源码为：

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

申明：newCachedThreadPool将建立一个可缓存的线程池，假如线程池的范围凌驾了处置惩罚需求时，那末将接纳余暇的线程，而当需求增添时，则增加新的线程，线程池的最大范围为Integer.MAX_VALUE。

然后看下newSingleThreadExecutor()要领的运用体式格局：

ExecutorService threadPool = Executors.newSingleThreadExecutor();

它的源码为：

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

申明：newSingleThreadExecutor是一个单线程的Executor，它建立单个事情者线程来实行使命，假如这个线程非常终了，就建立一个新的线程来替换。

newSingleThreadExecutor能够确保遵照使命在行列中的递次来串行实行。

末了看下newScheduledThreadPool()要领的运用体式格局：

ExecutorService threadPool = Executors.newScheduledThreadPool(10);

它的源码为：

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

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

super指向以下代码：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

申明：newScheduledThreadPool将建立一个牢固长度的线程池，而且以耽误或许定时的体式格局来实行使命，类似于Timer。

能够发明，以上4种体式格局终究都指向了ThreadPoolExecutor的以下组织函数，只是许多参数没让你指定，传递了默许值罢了：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    // 省略详细的代码
}

虽然运用这4个要领能够疾速的建立线程池，但照样不引荐运用，第一，许多参数都设置了默许值，不便于你明白各个参数的详细寄义，第二，参数的默许值可能会形成肯定的题目，最好是由运用者依据本身的需求自行指定。

那末这7个参数离别代表什么寄义呢？请接着往下看。

3.2 运用ThreadPoolExecutor组织函数建立（引荐）

ThreadPoolExecutor共有以下4个组织函数，引荐运用这类体式格局来建立线程池：

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

以上3个也都指向参数最全的第4个组织函数：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    // 省略详细的代码
}

以下为各个参数的解说：

corePoolSize：中心线程数。
maximumPoolSize：最大线程数。

最大线程数=中心线程数+非中心线程数。
keepAliveTime：非中心线程闲置超时时候。

一个非中心线程，假如不干活(闲置状况)的时长凌驾这个参数所设定的时长，就会被烧毁掉，假如设置了allowCoreThreadTimeOut = true，则会作用于中心线程。
unit：参数keepAliveTime的时候单元，如秒、分、小时。
workQueue：事情行列，即要实行的使命行列，内里存储守候实行的使命。

这里的壅塞行列可选择的有：LinkedBlockingQueue、ArrayBlockingQueue、SynchronousQueue、DelayedWorkQueue。

newFixedThreadPool()要领默许运用的LinkedBlockingQueue，

newCachedThreadPool()要领默许运用的SynchronousQueue，

newSingleThreadExecutor()要领默许运用的LinkedBlockingQueue，

newScheduledThreadPool()要领默许运用的DelayedWorkQueue。
threadFactory：线程工场，用来用来建立线程。
handler：饱和战略/谢绝处置惩罚使命时的战略。

当workQueue已满，而且线程池的线程数已抵达maximumPoolSize，此时新提交的使命会交由RejectedExecutionHandler handler处置惩罚，主要有以下4种战略：

AbortPolicy：中断战略，扬弃使命并抛出未搜检的RejectedExecutionException，这也是默许的饱和战略。

DiscardPolicy：扬弃战略，直接扬弃使命，但不抛出非常。

DiscardOldestPolicy：扬弃最旧的战略，扬弃下一个将被实行的使命，然后尝试从新提交新的使命。

CallerRunsPolicy：挪用者运转战略，将使命回退到挪用者，在挪用者地点的线程实行该使命。

4. 线程池的运转道理

能够经由过程下面2张图来明白线程池的运转道理：

1)假如线程池中的线程小于corePoolSize，则建立新线程来处置惩罚使命，这时候建立的线程为中心线程。

2)假如线程中的线程即是或许大于corePoolSize，则将使命放到事情行列中，即上图中的BlockingQueue。

3)假如事情行列已满，没法将使命加入到BlockingQueue，则建立新的线程来处置惩罚使命，这时候建立的线程为非中心线程，非中心线程在余暇一段时候后会被接纳烧毁掉（keepAliveTime和unit就是用来定义这个余暇的时候是若干）。

4)假如建立新线程致使线程池中的线程数凌驾了maximumPoolSize，使命将被谢绝，并挪用RejectedExecutionHandler.rejectedExecution()要领。

5. ThreadPoolExecutor示例

新建以下示例代码，建立1个corePoolSize为2，maximumPoolSize为3的线程池：

import java.util.concurrent.*;

public class ThreadPoolExecutorTest {
    public static void main(String[] args) {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 3, 60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1));

        threadPoolExecutor.execute(() -> {
            try {
                Thread.sleep(3 * 1000);
                System.out.println("使命1实行线程：" + Thread.currentThread().getName());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        threadPoolExecutor.execute(() -> {
            System.out.println("使命2实行线程：" + Thread.currentThread().getName());
        });
    }
}

运转效果为：

使命2实行线程：pool-1-thread-2

使命1实行线程：pool-1-thread-1

能够看出，因为线程池中的线程数小于corePoolSize，线程池建立了2个中心线程来离别实行使命1和使命2。

修正代码为以下所示，开启3个使命：

import java.util.concurrent.*;

public class ThreadPoolExecutorTest {
    public static void main(String[] args) {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 3, 60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1));

        threadPoolExecutor.execute(() -> {
            try {
                Thread.sleep(3 * 1000);
                System.out.println("使命1实行线程：" + Thread.currentThread().getName());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        threadPoolExecutor.execute(() -> {
            try {
                Thread.sleep(5 * 1000);
                System.out.println("使命2实行线程：" + Thread.currentThread().getName());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        threadPoolExecutor.execute(() -> System.out.println("使命3实行线程：" + Thread.currentThread().getName()));
    }
}

运转效果为：

使命1实行线程：pool-1-thread-1

使命3实行线程：pool-1-thread-1

使命2实行线程：pool-1-thread-2

能够看出，实行使命3时并没有新建线程，而是先放入了事情行列，末了由线程1实行完成。

在上面的代码中新增个使命4：

threadPoolExecutor.execute(() -> System.out.println("使命4实行线程：" + Thread.currentThread().getName()));

此时运转效果为：

使命4实行线程：pool-1-thread-3

使命3实行线程：pool-1-thread-3

使命1实行线程：pool-1-thread-1

使命2实行线程：pool-1-thread-2

能够看出，使命3是先放入了事情行列，使命4放不到事情行列（空间已满），所以建立了第3个线程来实行，实行终了后从行列里猎取到使命3实行，使命1和使命2离别由线程1和线程2实行。

修正下使命4的代码，并增加使命5：

threadPoolExecutor.execute(() -> {
    try {
        Thread.sleep(2 * 1000);
        System.out.println("使命4实行线程：" + Thread.currentThread().getName());
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
});

threadPoolExecutor.execute(() -> System.out.println("使命5实行线程：" + Thread.currentThread().getName()));

此时运转效果为：

Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task ThreadPoolExecutorTest$$Lambda$5/935044096@179d3b25 rejected from java.util.concurrent.ThreadPoolExecutor@254989ff[Running, pool size = 3, active threads = 3, queued tasks = 1, completed tasks = 0]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
at ThreadPoolExecutorTest.main(ThreadPoolExecutorTest.java:37)
使命4实行线程：pool-1-thread-3

使命3实行线程：pool-1-thread-3

使命1实行线程：pool-1-thread-1

使命2实行线程：pool-1-thread-2

能够看出，当提交使命5时，因为事情行列已满，且线程池中的线程数已为3，所以该使命被扬弃并抛出了java.util.concurrent.RejectedExecutionException非常。