Java｜线程池学习总结
#Java #多线程 [字体 小·中·大]一、常用接口及实现类
1. Executor、ExecutorService
Executor 里定义了一个 execute 方法，这个方法接受一个实现 Runnable 的对象。
1public interface Executor {
2 void execute(Runnable command);
3}
ExecutorService 是继承 Excutor 的一个接口，它定义了 Executor 框架常用的方法，如提交任务、线程池关闭、判断线程池的状态等方法。
2. Runnable、Callable
Runnable 定义了一个没有返回值的可执行方法，Callable
实现这两个接口的类都可以做为线程 Thread 执行的对象，这里演示一个 Callable。
1 public void test() throws Exception {
2 FutureTask<String> task = new FutureTask<>(new Callable() {
3 @Override
4 public Object call() throws Exception {
5 String res = "hello world";
6 Thread.sleep(3000);
7 return res;
8 }
9 });
10
11 new Thread(task).start();
12
13 task.get();//call return前阻塞
14 }
1 public static void main(String[] args) throws ExecutionException, InterruptedException {
2 Callable<String> c = new Callable() {
3 @Override
4 public String call() throws Exception {
5 return "Hello Callable";
6 }
7 };
8
9 ExecutorService service = Executors.newCachedThreadPool();
10 Future<String> future = service.submit(c); //异步
11
12 System.out.println(future.get());//获取到结果前阻塞
13
14 service.shutdown();
15 }
3. Executors 线程池工具
Executors 是线程池的工具类，类似于集合框架的 Collections 工具类。
已经预定义了一些线程池，设置参数后可以直接使用。
4. 线程池的关闭
线程不是立即关闭（终结）的，关闭和终结是两个不同的状态。
线程池的关闭即遍历每个线程然后触发线程的中断（interrupt）。
1 public static void main(String[] args) throws InterruptedException {
2 ExecutorService service = Executors.newFixedThreadPool(5); //execute submit
3 for (int i = 0; i < 6; i++) {
4 service.execute(() -> {
5 try {
6 TimeUnit.MILLISECONDS.sleep(500);
7 } catch (InterruptedException e) {
8 e.printStackTrace();
9 }
10 System.out.println(Thread.currentThread().getName());
11 });
12 }
13 System.out.println(service);
14
15 // 关闭线程
16 service.shutdown();
17 // isTerminated() 为false说明线程run方法还未执行完毕
18 System.out.println(service.isTerminated());
19 //此时线程已经处在关闭状态，但是还未终结（终结是线程执行完毕从run方法退出）
20 System.out.println(service.isShutdown());
21 System.out.println(service);
22
23 // 在休眠会查看
24 TimeUnit.SECONDS.sleep(5);
25 // true 所有线程已终结
26 System.out.println(service.isTerminated());
27 // true 所有线程已关闭
28 System.out.println(service.isShutdown());
29 System.out.println(service);
30 }
31
32// output
33java.util.concurrent.ThreadPoolExecutor@76ccd017[Running, pool size = 5, active threads = 5, queued tasks = 1, completed tasks = 0]
34false
35true
36java.util.concurrent.ThreadPoolExecutor@76ccd017[Shutting down, pool size = 5, active threads = 5, queued tasks = 1, completed tasks = 0]
37pool-1-thread-1
38pool-1-thread-4
39pool-1-thread-3
40pool-1-thread-2
41pool-1-thread-5
42pool-1-thread-1
43true
44true
45java.util.concurrent.ThreadPoolExecutor@76ccd017[Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 6]
5. Futrue 异步任务
Future 是异步作业，它表示提交作业未来的一个计算结果，在使用 get 方法获取结果时如果 call()未返回结果则会阻塞当前线程。
FutrueTask 是 Futrue 的实现类，可以接受 Runnable、Callable 实现类为一个执行作业。
1 public static void main(String[] args) throws InterruptedException, ExecutionException {
2
3 FutureTask<Integer> task = new FutureTask<>(()->{
4 TimeUnit.MILLISECONDS.sleep(500);
5 return 1000;
6 }); //new Callable () { Integer call();}
7
8 new Thread(task).start();
9
10 System.out.println(task.get()); //阻塞，知道call返回结果
11 }
5.1 CompletableFuture
CompletableFuture 是一个处理异步任务的线程工具，使用它可以对异步任务进行控制或者继续进行之后的操作（这点类似 Stream）。
下面是个使用示例。
1/**
2 * 假设能够提供一个服务
3 * 这个服务查询各大电商网站同一类产品的价格并汇总展示
4 * 可以使用单线程一个一个查，也可以使用异步并行方式查
5 */
6
7import java.io.IOException;
8import java.util.concurrent.CompletableFuture;
9import java.util.concurrent.ExecutionException;
10import java.util.concurrent.TimeUnit;
11
12public class TestCompletableFuture {
13 public static void main(String[] args) throws ExecutionException, InterruptedException {
14 long start, end;
15 // 方案1，单线程计算
16 /*start = System.currentTimeMillis();
17
18 priceOfTM();
19 priceOfTB();
20 priceOfJD();
21
22 end = System.currentTimeMillis();
23 System.out.println("use serial method call! " + (end - start));*/
24
25 // ----------------------------------------------------------------------------------
26
27 // 方案2，异步并行计算
28 start = System.currentTimeMillis();
29
30 // 提交异步任务
31 CompletableFuture<Double> futureTM = CompletableFuture.supplyAsync(()->priceOfTM());
32 CompletableFuture<Double> futureTB = CompletableFuture.supplyAsync(()->priceOfTB());
33 CompletableFuture<Double> futureJD = CompletableFuture.supplyAsync(()->priceOfJD());
34
35 // 等待作业完成
36 CompletableFuture.allOf(futureTM, futureTB, futureJD).join();
37
38 // CompletableFuture 还可以链式进行其他操作
39 /*CompletableFuture.supplyAsync(()->priceOfTM())
40 .thenApply(String::valueOf)
41 .thenApply(str-> "price " + str)
42 .thenAccept(System.out::println);*/
43
44 end = System.currentTimeMillis();
45 System.out.println("use completable future! " + (end - start));
46
47 // 阻塞等待结果
48 try {
49 System.in.read();
50 } catch (IOException e) {
51 e.printStackTrace();
52 }
53 }
54
55 private static double priceOfTM() {
56 delay();
57 return 1.00;
58 }
59
60 private static double priceOfTB() {
61 delay();
62 return 2.00;
63 }
64
65 private static double priceOfJD() {
66 delay();
67 return 3.00;
68 }
69
70 // 延迟工具类
71 private static void delay() {
72// int time = new Random().nextInt(500);
73 int time = 150;
74 try {
75 TimeUnit.MILLISECONDS.sleep(time);
76 } catch (InterruptedException e) {
77 e.printStackTrace();
78 }
79 System.out.printf("After %s sleep!\n", time);
80 }
81}
二、Executors
Executors 是线程工具类，提供了一套开箱即用的线程池，只需要简单设置参数即可使用。
1. SingleThreadExecutor
SingleThreadExecutor 单线程线程池, 只会有一个线程，少于一个线程会创建一个线程。
使用
1 public static void main(String[] args) {
2 ExecutorService service = Executors.newSingleThreadExecutor();
3 for(int i=0; i<5; i++) {
4 final int j = i;
5 service.execute(()->{
6 System.out.println(j + " " + Thread.currentThread().getName());
7 });
8 }
9 }
源码
1 public static ExecutorService newSingleThreadExecutor() {
2 return new FinalizableDelegatedExecutorService
3 // 核心线程数1，最大线程数1
4 (new ThreadPoolExecutor(1, 1,
5 0L, TimeUnit.MILLISECONDS,
6 new LinkedBlockingQueue<Runnable>()));
7 }
2. 缓存线程池
使用
1 public static void main(String[] args) throws InterruptedException {
2 ExecutorService service = Executors.newCachedThreadPool();
3 // 刚初始化，线程数0
4 System.out.println(service);
5
6 for (int i = 0; i < 2; i++) {
7 service.execute(() -> {
8 try {
9 TimeUnit.MILLISECONDS.sleep(500);
10 } catch (InterruptedException e) {
11 e.printStackTrace();
12 }
13 System.out.println(Thread.currentThread().getName());
14 });
15 }
16 //size=2 线程数2,active threads=2，活跃线程数2
17 System.out.println(service);
18
19 //60秒内无作业处理线程就会被关闭
20 TimeUnit.SECONDS.sleep(80);
21
22 //size=0 线程数0，active threads=0 活跃线程0
23 System.out.println(service);
24
25 }
源码
1 public static ExecutorService newCachedThreadPool() {
2 return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
3 //活跃时间60秒
4 60L, TimeUnit.SECONDS,
5 new SynchronousQueue<Runnable>());
6 }
2. FixedThreadPool
固定线程池一直保持固定个线程活跃。
使用
1public class T09_FixedThreadPool {
2 public static void main(String[] args) throws InterruptedException, ExecutionException {
3 // 单线程处理判断20W个数是否为质数
4 // 测试1，单线程
5 long start = System.currentTimeMillis();
6 getPrime(1, 200000);
7 long end = System.currentTimeMillis();
8 System.out.println("单线程耗时:"+(end - start));
9
10 // 测试2，多线程，线程池，4个线程
11 final int cpuCoreNum = 4;
12
13 ExecutorService service = Executors.newFixedThreadPool(cpuCoreNum);
14
15 // 20W个数分四个作业
16 MyTask t1 = new MyTask(1, 80000); //1-5 5-10 10-15 15-20
17 MyTask t2 = new MyTask(80001, 130000);
18 MyTask t3 = new MyTask(130001, 170000);
19 MyTask t4 = new MyTask(170001, 200000);
20
21 // 计时
22 start = System.currentTimeMillis();
23
24 // 提交作业
25 Future<List<Integer>> f1 = service.submit(t1);
26 Future<List<Integer>> f2 = service.submit(t2);
27 Future<List<Integer>> f3 = service.submit(t3);
28 Future<List<Integer>> f4 = service.submit(t4);
29
30 // 阻塞执行完成
31 f1.get();
32 f2.get();
33 f3.get();
34 f4.get();
35 end = System.currentTimeMillis();
36 System.out.println("线程池耗时:"+(end - start));
37 }
38
39 static class MyTask implements Callable<List<Integer>> {
40 int startPos, endPos;
41
42 MyTask(int s, int e) {
43 this.startPos = s;
44 this.endPos = e;
45 }
46
47 @Override
48 public List<Integer> call() throws Exception {
49 List<Integer> r = getPrime(startPos, endPos);
50 return r;
51 }
52
53 }
54
55 static boolean isPrime(int num) {
56 for(int i=2; i<=num/2; i++) {
57 if(num % i == 0) return false;
58 }
59 return true;
60 }
61
62 static List<Integer> getPrime(int start, int end) {
63 List<Integer> results = new ArrayList<>();
64 for(int i=start; i<=end; i++) {
65 if(isPrime(i)) results.add(i);
66 }
67
68 return results;
69 }
70}
源码
1 public static ExecutorService newFixedThreadPool(int nThreads) {
2 return new ThreadPoolExecutor(nThreads, nThreads,
3 0L, TimeUnit.MILLISECONDS,
4 new LinkedBlockingQueue<Runnable>());
5 }
4. ScheduledThreadPoolExecutor
定时任务线程池，指定固定个线程，延迟或重复执行作业。
使用
1 // 设置一个定时任务，每2秒执行一次作业处理
2 public static void main(String[] args) {
3 ScheduledExecutorService service = Executors.newScheduledThreadPool(4);
4 service.scheduleAtFixedRate(() -> {
5 int s = new Random().nextInt(1000);
6 try {
7 TimeUnit.MILLISECONDS.sleep(s);
8 } catch (InterruptedException e) {
9 e.printStackTrace();
10 }
11 System.out.println(Thread.currentThread().getName() + " 耗时 " + s + " ms 处理了作业");
12 }, 0, 2, TimeUnit.SECONDS);
13 }
5. WorkStealingPool
newWorkStealingPool，这个是 JDK1.8 版本加入的一种线程池，stealing 翻译为抢断、窃取的意思。
特点：
- 抢占工作
- 作业无序执行
源码
1 // 使用的是WorkJoinPool，与上边的几个不同
2 public static ExecutorService newWorkStealingPool() {
3 return new ForkJoinPool
4 (Runtime.getRuntime().availableProcessors(),
5 ForkJoinPool.defaultForkJoinWorkerThreadFactory,
6 null, true);
7 }
5.1 ForkJoinPool
https://zhuanlan.zhihu.com/p/90958193
三、ThreadPoolExecutor 源码分析
1. 常用变量的解释
1// 1. `ctl`，可以看做一个int类型的数字，高3位表示线程池状态，低29位表示worker数量
2private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
3// 2. `COUNT_BITS`，`Integer.SIZE`为32，所以`COUNT_BITS`为29
4private static final int COUNT_BITS = Integer.SIZE - 3;
5// 3. `CAPACITY`，线程池允许的最大线程数。1左移29位，然后减1，即为 2^29 - 1
6private static final int CAPACITY = (1 << COUNT_BITS) - 1;
7
8// runState is stored in the high-order bits
9// 4. 线程池有5种状态，按大小排序如下：RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED
10private static final int RUNNING = -1 << COUNT_BITS;
11private static final int SHUTDOWN = 0 << COUNT_BITS;
12private static final int STOP = 1 << COUNT_BITS;
13private static final int TIDYING = 2 << COUNT_BITS;
14private static final int TERMINATED = 3 << COUNT_BITS;
15
16// Packing and unpacking ctl
17// 5. `runStateOf()`，获取线程池状态，通过按位与操作，低29位将全部变成0
18private static int runStateOf(int c) { return c & ~CAPACITY; }
19// 6. `workerCountOf()`，获取线程池worker数量，通过按位与操作，高3位将全部变成0
20private static int workerCountOf(int c) { return c & CAPACITY; }
21// 7. `ctlOf()`，根据线程池状态和线程池worker数量，生成ctl值
22private static int ctlOf(int rs, int wc) { return rs | wc; }
23
24/*
25 * Bit field accessors that don't require unpacking ctl.
26 * These depend on the bit layout and on workerCount being never negative.
27 */
28// 8. `runStateLessThan()`，线程池状态小于xx
29private static boolean runStateLessThan(int c, int s) {
30 return c < s;
31}
32// 9. `runStateAtLeast()`，线程池状态大于等于xx
33private static boolean runStateAtLeast(int c, int s) {
34 return c >= s;
35}
2. 构造方法
1public ThreadPoolExecutor(int corePoolSize,
2 int maximumPoolSize,
3 long keepAliveTime,
4 TimeUnit unit,
5 BlockingQueue<Runnable> workQueue,
6 ThreadFactory threadFactory,
7 RejectedExecutionHandler handler) {
8 // 基本类型参数校验
9 if (corePoolSize < 0 ||
10 maximumPoolSize <= 0 ||
11 maximumPoolSize < corePoolSize ||
12 keepAliveTime < 0)
13 throw new IllegalArgumentException();
14 // 空指针校验
15 if (workQueue == null || threadFactory == null || handler == null)
16 throw new NullPointerException();
17 this.corePoolSize = corePoolSize;
18 this.maximumPoolSize = maximumPoolSize;
19 this.workQueue = workQueue;
20 // 根据传入参数`unit`和`keepAliveTime`，将存活时间转换为纳秒存到变量`keepAliveTime `中
21 this.keepAliveTime = unit.toNanos(keepAliveTime);
22 this.threadFactory = threadFactory;
23 this.handler = handler;
24}
3. 提交执行 task 的过程
1public void execute(Runnable command) {
2 if (command == null)
3 throw new NullPointerException();
4 /*
5 * Proceed in 3 steps:
6 *
7 * 1. If fewer than corePoolSize threads are running, try to
8 * start a new thread with the given command as its first
9 * task. The call to addWorker atomically checks runState and
10 * workerCount, and so prevents false alarms that would add
11 * threads when it shouldn't, by returning false.
12 *
13 * 2. If a task can be successfully queued, then we still need
14 * to double-check whether we should have added a thread
15 * (because existing ones died since last checking) or that
16 * the pool shut down since entry into this method. So we
17 * recheck state and if necessary roll back the enqueuing if
18 * stopped, or start a new thread if there are none.
19 *
20 * 3. If we cannot queue task, then we try to add a new
21 * thread. If it fails, we know we are shut down or saturated
22 * and so reject the task.
23 */
24 int c = ctl.get();
25 // worker数量比核心线程数小，直接创建worker执行任务
26 if (workerCountOf(c) < corePoolSize) {
27 if (addWorker(command, true))
28 return;
29 c = ctl.get();
30 }
31 // worker数量超过核心线程数，任务直接进入队列
32 if (isRunning(c) && workQueue.offer(command)) {
33 int recheck = ctl.get();
34 // 线程池状态不是RUNNING状态，说明执行过shutdown命令，需要对新加入的任务执行reject()操作。
35 // 这儿为什么需要recheck，是因为任务入队列前后，线程池的状态可能会发生变化。
36 if (! isRunning(recheck) && remove(command))
37 reject(command);
38 // 这儿为什么需要判断0值，主要是在线程池构造方法中，核心线程数允许为0
39 else if (workerCountOf(recheck) == 0)
40 addWorker(null, false);
41 }
42 // 如果线程池不是运行状态，或者任务进入队列失败，则尝试创建worker执行任务。
43 // 这儿有3点需要注意：
44 // 1. 线程池不是运行状态时，addWorker内部会判断线程池状态
45 // 2. addWorker第2个参数表示是否创建核心线程
46 // 3. addWorker返回false，则说明任务执行失败，需要执行reject操作
47 else if (!addWorker(command, false))
48 reject(command);
49}
4. addworker 源码解析
1private boolean addWorker(Runnable firstTask, boolean core) {
2 retry:
3 // 外层自旋
4 for (;;) {
5 int c = ctl.get();
6 int rs = runStateOf(c);
7
8 // 这个条件写得比较难懂，我对其进行了调整，和下面的条件等价
9 // (rs > SHUTDOWN) ||
10 // (rs == SHUTDOWN && firstTask != null) ||
11 // (rs == SHUTDOWN && workQueue.isEmpty())
12 // 1. 线程池状态大于SHUTDOWN时，直接返回false
13 // 2. 线程池状态等于SHUTDOWN，且firstTask不为null，直接返回false
14 // 3. 线程池状态等于SHUTDOWN，且队列为空，直接返回false
15 // Check if queue empty only if necessary.
16 if (rs >= SHUTDOWN &&
17 ! (rs == SHUTDOWN &&
18 firstTask == null &&
19 ! workQueue.isEmpty()))
20 return false;
21
22 // 内层自旋
23 for (;;) {
24 int wc = workerCountOf(c);
25 // worker数量超过容量，直接返回false
26 if (wc >= CAPACITY ||
27 wc >= (core ? corePoolSize : maximumPoolSize))
28 return false;
29 // 使用CAS的方式增加worker数量。
30 // 若增加成功，则直接跳出外层循环进入到第二部分
31 if (compareAndIncrementWorkerCount(c))
32 break retry;
33 c = ctl.get(); // Re-read ctl
34 // 线程池状态发生变化，对外层循环进行自旋
35 if (runStateOf(c) != rs)
36 continue retry;
37 // 其他情况，直接内层循环进行自旋即可
38 // else CAS failed due to workerCount change; retry inner loop
39 }
40 }
41 boolean workerStarted = false;
42 boolean workerAdded = false;
43 Worker w = null;
44 try {
45 w = new Worker(firstTask);
46 final Thread t = w.thread;
47 if (t != null) {
48 final ReentrantLock mainLock = this.mainLock;
49 // worker的添加必须是串行的，因此需要加锁
50 mainLock.lock();
51 try {
52 // Recheck while holding lock.
53 // Back out on ThreadFactory failure or if
54 // shut down before lock acquired.
55 // 这儿需要重新检查线程池状态
56 int rs = runStateOf(ctl.get());
57
58 if (rs < SHUTDOWN ||
59 (rs == SHUTDOWN && firstTask == null)) {
60 // worker已经调用过了start()方法，则不再创建worker
61 if (t.isAlive()) // precheck that t is startable
62 throw new IllegalThreadStateException();
63 // worker创建并添加到workers成功
64 workers.add(w);
65 // 更新`largestPoolSize`变量
66 int s = workers.size();
67 if (s > largestPoolSize)
68 largestPoolSize = s;
69 workerAdded = true;
70 }
71 } finally {
72 mainLock.unlock();
73 }
74 // 启动worker线程
75 if (workerAdded) {
76 t.start();
77 workerStarted = true;
78 }
79 }
80 } finally {
81 // worker线程启动失败，说明线程池状态发生了变化（关闭操作被执行），需要进行shutdown相关操作
82 if (! workerStarted)
83 addWorkerFailed(w);
84 }
85 return workerStarted;
86}
5. 线程池 worker 任务单元
1private final class Worker
2 extends AbstractQueuedSynchronizer
3 implements Runnable
4{
5 /**
6 * This class will never be serialized, but we provide a
7 * serialVersionUID to suppress a javac warning.
8 */
9 private static final long serialVersionUID = 6138294804551838833L;
10
11 /** Thread this worker is running in. Null if factory fails. */
12 final Thread thread;
13 /** Initial task to run. Possibly null. */
14 Runnable firstTask;
15 /** Per-thread task counter */
16 volatile long completedTasks;
17
18 /**
19 * Creates with given first task and thread from ThreadFactory.
20 * @param firstTask the first task (null if none)
21 */
22 Worker(Runnable firstTask) {
23 setState(-1); // inhibit interrupts until runWorker
24 this.firstTask = firstTask;
25 // 这儿是Worker的关键所在，使用了线程工厂创建了一个线程。传入的参数为当前worker
26 this.thread = getThreadFactory().newThread(this);
27 }
28
29 /** Delegates main run loop to outer runWorker */
30 public void run() {
31 runWorker(this);
32 }
33
34 // 省略代码...
35}
6. 核心线程执行逻辑-runworker
1final void runWorker(Worker w) {
2 Thread wt = Thread.currentThread();
3 Runnable task = w.firstTask;
4 w.firstTask = null;
5 // 调用unlock()是为了让外部可以中断
6 w.unlock(); // allow interrupts
7 // 这个变量用于判断是否进入过自旋（while循环）
8 boolean completedAbruptly = true;
9 try {
10 // 这儿是自旋
11 // 1. 如果firstTask不为null，则执行firstTask；
12 // 2. 如果firstTask为null，则调用getTask()从队列获取任务。
13 // 3. 阻塞队列的特性就是：当队列为空时，当前线程会被阻塞等待
14 while (task != null || (task = getTask()) != null) {
15 // 这儿对worker进行加锁，是为了达到下面的目的
16 // 1. 降低锁范围，提升性能
17 // 2. 保证每个worker执行的任务是串行的
18 w.lock();
19 // If pool is stopping, ensure thread is interrupted;
20 // if not, ensure thread is not interrupted. This
21 // requires a recheck in second case to deal with
22 // shutdownNow race while clearing interrupt
23 // 如果线程池正在停止，则对当前线程进行中断操作
24 if ((runStateAtLeast(ctl.get(), STOP) ||
25 (Thread.interrupted() &&
26 runStateAtLeast(ctl.get(), STOP))) &&
27 !wt.isInterrupted())
28 wt.interrupt();
29 // 执行任务，且在执行前后通过`beforeExecute()`和`afterExecute()`来扩展其功能。
30 // 这两个方法在当前类里面为空实现。
31 try {
32 beforeExecute(wt, task);
33 Throwable thrown = null;
34 try {
35 task.run();
36 } catch (RuntimeException x) {
37 thrown = x; throw x;
38 } catch (Error x) {
39 thrown = x; throw x;
40 } catch (Throwable x) {
41 thrown = x; throw new Error(x);
42 } finally {
43 afterExecute(task, thrown);
44 }
45 } finally {
46 // 帮助gc
47 task = null;
48 // 已完成任务数加一
49 w.completedTasks++;
50 w.unlock();
51 }
52 }
53 completedAbruptly = false;
54 } finally {
55 // 自旋操作被退出，说明线程池正在结束
56 processWorkerExit(w, completedAbruptly);
57 }
58}