文章目录
- 1. 线程池状态
- 2. 线程池创建
- 2.1 ThreadPoolExecutor
- 2.2 newFixedThreadPool
- 2.3 newCachedThreadPool
- 2.4 newSingleThreadExecutor
- 2.5 开发要求
- 3. 提交方法
- 4. 关闭方法
- 5. 任务调度
- 5.1 Timer
- 5.2 Scheduled
- 5.2.1 构造方法
- 5.2.2 schedule
- 5.2.3 scheduleAtFixedRate
- 5.2.4 scheduleWithFixedDelay
- 5.3 Tomcat线程池
- 6. Fork/Join
JAVA的多线程绕不过的就是线程池了，在实际的开发中，我们并没有一个线程一个线程的创建再去执行任务，这样的话势必会频繁的开启和关闭线程，极大的影响程序的效率。由此，我们一般使用的线程池来进行程序的开发。
线程池实际就是一个管理几个线程的管理器，当有任务来的时候，如果线程池中有闲置的线程，那么就使用闲置的线程来进行任务的处理，如果没有闲置的线程，那么有很多种处理方式，比如抛出错误、等待任务执行等等，这些在后文会提到。
首先，我们先来看看线程池的各个状态以及各个状态在Java中的表示。
1. 线程池状态
| 状态名 | 高3位 | 接收新任务 | 处理阻塞任务 | 说明 |
|---|---|---|---|---|
| RUNNING | 111 | Y | Y | |
| SHUTDOWN | 000 | N | Y | 不会接收新任务，但会处理阻塞队列剩余任务 |
| STOP | 001 | N | N | 会中断正在执行的任务，并抛弃阻塞队列任务 |
| TIDYING | 010 | 任务全执行完毕，活动线程为 0 即将进入终结 | ||
| TERMINATED | 011 | 终结状态 |
从数字上比较，TERMINATED > TIDYING > STOP > SHUTDOWN > RUNNING，因为最高位表示符号位。
这些信息存储在一个原子变量 ctl 中，目的是将线程池状态与线程个数合二为一，这样就可以用一次 cas 原子操作进行赋值。
// c 为旧值， ctlOf 返回结果为新值
ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))));
// rs 为高 3 位代表线程池状态， wc 为低 29 位代表线程个数，ctl 是合并它们
private static int ctlOf(int rs, int wc) { return rs | wc; }
2. 线程池创建
2.1 ThreadPoolExecutor
ThreadPoolExecutor的构造方法如下：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
各参数含义如下：
| 参数 | 含义 |
|---|---|
corePoolSize | 核心线程数目 (最多保留的线程数) |
maximumPoolSize | 最大线程数目 |
keepAliveTime | 生存时间 - 针对救急线程 |
unit | 针对救急线程 |
workQueue | 阻塞队列 |
threadFactory | 线程工厂 - 可以为线程创建时起个好名字 |
handler | 拒绝策略，即线程池满时再有任务来会怎么处理 |
其工作流程如下：
-
线程池中刚开始没有线程，当一个任务提交给线程池后，线程池会创建一个新线程来执行任务。
-
当线程数达到
corePoolSize并没有线程空闲，这时再加入任务，新加的任务会被加入workQueue队列排队，直到有空闲的线程。 -
如果队列选择了有界队列，那么任务超过了队列大小时，会创建
maximumPoolSize - corePoolSize数目的线程来救急。 -
如果线程到达
maximumPoolSize仍然有新任务这时会执行拒绝策略。拒绝策略 jdk 提供了 4 种实现，其它著名框架也提供了实现- AbortPolicy 让调用者抛出
RejectedExecutionException异常，这是默认策略 - CallerRunsPolicy 让调用者运行任务
- DiscardPolicy 会放弃本次任务
- DiscardOldestPolicy 放弃队列中最早的任务，本任务取而代之
其他框架的拒绝策略如下：
- Dubbo 的实现，在抛出
RejectedExecutionException异常之前会记录日志，并 dump 线程栈信息，方便定位问题 - Netty 的实现，是创建一个新线程来执行任务
- ActiveMQ 的实现，带超时等待（60s）尝试放入队列，类似我们之前自定义的拒绝策略
- PinPoint 的实现，它使用了一个拒绝策略链，会逐一尝试策略链中每种拒绝策略
- AbortPolicy 让调用者抛出
-
当高峰过去后，超过
corePoolSize的救急线程如果一段时间没有任务做，需要结束节省资源，这个时间由keepAliveTime和unit来控制。
根据ThreadPoolExecutor这个构造方法，JDK Executors 类中提供了众多工厂方法来创建各种用途的线程池。
2.2 newFixedThreadPool
newFixedThreadPool的构造方法如下：
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
特点：
- 核心线程数 == 最大线程数（没有救急线程被创建），因此也无需超时时间
- LinkedBlockingQueue 是一个单向链表实现的阻塞队列，默认大小为
Integer.MAX_VALUE，也就是无界队列，可以放任意数量的任务，在任务比较多的时候会导致 OOM（内存溢出） - 适用于任务量已知，相对耗时的长期任务
2.3 newCachedThreadPool
newCachedThreadPool的构造方法如下：
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
特点：
- 核心线程数是 0， 最大线程数是 Integer.MAX_VALUE，救急线程的空闲生存时间是 60s，意味着全部都是救急线程（60s 后可以回收），并且救急线程可以无限创建
- SynchronousQueue 作为阻塞队列，没有容量，对于每一个 take 的线程会阻塞直到有一个 put 的线程放入元素为止（类似一手交钱、一手交货）
- 适合任务数比较密集，但每个任务执行时间较短的情况
2.4 newSingleThreadExecutor
newSingleThreadExecutor构造方法如下：
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
特点：
- 保证所有任务按照指定顺序执行，线程数固定为 1，任务数多于 1 时会放入无界队列排队，任务执行完毕，这唯一的线程也不会被释放
- 自己创建一个单线程串行执行任务，如果任务执行失败而终止那么没有任何补救措施，而线程池还会新建一个线程，保证池的正常工作
与Executors.newFixedThreadPool(1)的区别：
- FinalizableDelegatedExecutorService 应用的是装饰器模式，只对外暴露了 ExecutorService 接口，因此不能调用 ThreadPoolExecutor 中特有的方法
- 对外暴露的是 ThreadPoolExecutor 对象，可以强转后调用 setCorePoolSize 等方法进行修改
2.5 开发要求
阿里巴巴 Java 开发手册要求：
-
线程资源必须通过线程池提供，不允许在应用中自行显式创建线程
- 使用线程池的好处是减少在创建和销毁线程上所消耗的时间以及系统资源的开销，解决资源不足的问题
- 如果不使用线程池，有可能造成系统创建大量同类线程而导致消耗完内存或者过度切换的问题
-
线程池不允许使用 Executors 去创建，而是通过 ThreadPoolExecutor 的方式，这样的处理方式更加明确线程池的运行规则，规避资源耗尽的风险
Executors 返回的线程池对象弊端如下：
- FixedThreadPool 和 SingleThreadPool：请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM
- CacheThreadPool 和 ScheduledThreadPool：允许创建线程数量为 Integer.MAX_VALUE，可能会创建大量的线程，导致 OOM
那么，创建多少线程合适呢？
对于CPU 密集型运算 ：
- 通常采用 cpu 核数 + 1 能够实现最优的 CPU 利用率，+1 是保证当线程由于页缺失故障（操作系统）或其它原因导致暂停时，额外的这个线程就能顶上去，保证 CPU 时钟周期不被浪费
对于I/O 密集型运算：
- CPU 不总是处于繁忙状态，例如，当你执行业务计算时，这时候会使用 CPU 资源，但当你执行 I/O 操作时、远程RPC 调用时，包括进行数据库操作时，这时候 CPU 就闲下来了，你可以利用多线程提高它的利用率，经验公式如下：
线程数 = 核数 ∗ 期望 C P U 利用率 ∗ 总时间 ( C P U 计算时间 + 等待时间 ) / C P U 计算时间 线程数 = 核数 * 期望 CPU 利用率 * 总时间(CPU计算时间+等待时间) / CPU 计算时间 线程数=核数∗期望CPU利用率∗总时间(CPU计算时间+等待时间)/CPU计算时间
例如 4 核 CPU 计算时间是 50% ，其它等待时间是 50%，期望 cpu 被 100% 利用，套用公式:
4 × 100 % × 100 % / 50 % = 8 4 \times 100\% \times 100\% / 50\% = 8 4×100%×100%/50%=8
3. 提交方法
| 方法 | 说明 |
|---|---|
void execute(Runnable command) | 执行任务（Executor 类 API） |
Future<?> submit(Runnable task) | 提交任务 task() |
Future submit(Callable<T> task) | 提交任务 task，用返回值 Future 获得任务执行结果 |
List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) | 提交 tasks 中所有任务 |
List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) | 提交 tasks 中所有任务，超时时间针对所有task，超时会取消没有执行完的任务，并抛出超时异常 |
T invokeAny(Collection<? extends Callable<T>> tasks) | 提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消 |
代码示例如下：
@Slf4j(topic = "c.Test20")
public class Test1 {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService pool = Executors.newFixedThreadPool(1);
// submit示例
// method1(pool);
// excute示例
// method2(pool);
// invokeAll示例
// method3(pool);
// invokeAny示例
method4(pool);
}
private static void method1(ExecutorService pool) throws ExecutionException, InterruptedException {
Future<String> running = pool.submit(() -> {
log.debug("running");
sleep(1000);
return "ok";
});
log.debug("{}", running.get());
}
private static void method2(ExecutorService pool) {
pool.execute(() -> {
log.debug("running");
try {
sleep(1000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
log.debug("finished");
});
}
private static void method3(ExecutorService pool) throws InterruptedException {
List<Future<String>> futures = pool.invokeAll(Arrays.asList(
() -> {
log.debug("begin1");
Thread.sleep(1000);
return "1";
},
() -> {
log.debug("begin2");
Thread.sleep(500);
return "2";
},
() -> {
log.debug("begin3");
Thread.sleep(2000);
return "3";
}
));
// 逐个获取返回值
futures.forEach( f -> {
try {
log.debug("{}", f.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
});
}
private static void method4(ExecutorService pool) throws ExecutionException, InterruptedException {
String result = pool.invokeAny(Arrays.asList(
() -> {
log.debug("begin 1");
Thread.sleep(1000);
log.debug("end 1");
return "1";
},
() -> {
log.debug("begin 2");
Thread.sleep(500);
log.debug("end 2");
return "2";
},
() -> {
log.debug("begin 3");
Thread.sleep(2000);
log.debug("end 3");
return "3";
}
));
// invokeAny只返回第一个执行完的线程的返回值
log.debug("{}", result);
}
}
execute 和 submit 都属于线程池的方法，对比：
-
execute 只能执行 Runnable 类型的任务，没有返回值； submit 既能提交 Runnable 类型任务也能提交 Callable 类型任务，底层是封装成 FutureTask，然后调用 execute 执行
-
execute 会直接抛出任务执行时的异常，submit 会吞掉异常，可通过 Future 的 get 方法将任务执行时的异常重新抛出
execute 会直接抛出任务执行时的异常，submit 会吞掉异常，有两种处理方法
方法 1：主动捉异常
ExecutorService pool = Executors.newFixedThreadPool(1);
pool.submit(() -> {
try {
System.out.println("task1");
int i = 1 / 0;
} catch (Exception e) {
e.printStackTrace();
}
});
方法 2：使用 Future 对象
ExecutorService pool = Executors.newFixedThreadPool(1);
Future<?> future = pool.submit(() -> {
System.out.println("task1");
int i = 1 / 0;
return true;
});
System.out.println(future.get());
4. 关闭方法
| 方法 | 说明 |
|---|---|
void shutdown() | 线程池状态变为 SHUTDOWN，等待任务执行完后关闭线程池，不会接收新任务，但已提交任务会执行完，而且也可以添加线程（不绑定任务） |
List<Runnable> shutdownNow() | 线程池状态变为 STOP，用 interrupt 中断正在执行的任务，直接关闭线程池，不会接收新任务，会将队列中的任务返回 |
boolean isShutdown() | 不在 RUNNING 状态的线程池，此执行者已被关闭，方法返回 true |
boolean isTerminated() | 线程池状态是否是 TERMINATED，如果所有任务在关闭后完成，返回 true |
boolean awaitTermination(long timeout, TimeUnit unit) | 调用 shutdown 后，由于调用线程不会等待所有任务运行结束，如果它想在线程池 TERMINATED 后做些事情，可以利用此方法等待 |
代码示例如下：
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService pool = Executors.newFixedThreadPool(2);
Future<Integer> result1 = pool.submit(() -> {
log.debug("task 1 running...");
Thread.sleep(1000);
log.debug("task 1 finish...");
return 1;
});
Future<Integer> result2 = pool.submit(() -> {
log.debug("task 2 running...");
Thread.sleep(1000);
log.debug("task 2 finish...");
return 2;
});
log.debug("shutdown");
pool.shutdown();
log.debug("other....");
}
上面代码的输出如下：
09:56:40.237 c.Test20 [main] - shutdown
09:56:40.237 c.Test20 [pool-1-thread-2] - task 2 running…
09:56:40.237 c.Test20 [pool-1-thread-1] - task 1 running…
09:56:40.240 c.Test20 [main] - other…
09:56:41.254 c.Test20 [pool-1-thread-2] - task 2 finish…
09:56:41.255 c.Test20 [pool-1-thread-1] - task 1 finish…
可以看到调用了shutdown后其立刻输出的是other，然后再执行的队列中的任务，如果我们需要等待任务执行完再输出other的话，可以使用awaitTermination语句：
log.debug("shutdown");
pool.shutdown();
pool.awaitTermination(3, TimeUnit.SECONDS);
log.debug("other....");
输出如下：
09:59:43.845 c.Test20 [pool-1-thread-1] - task 1 running…
09:59:43.845 c.Test20 [main] - shutdown
09:59:43.845 c.Test20 [pool-1-thread-2] - task 2 running…
09:59:44.859 c.Test20 [pool-1-thread-2] - task 2 finish…
09:59:44.859 c.Test20 [pool-1-thread-1] - task 1 finish…
09:59:44.859 c.Test20 [main] - other…
5. 任务调度
5.1 Timer
Timer 实现定时功能，Timer 的优点在于简单易用，但其缺点有很多：
- 但由于所有任务都是由同一个线程来调度，因此所有任务都是串行执行的，同一时间只能有一个任务在执行，前一个任务的延迟或异常都将会影响到之后的任务
- 如果前一个任务报错了，那么后面的任务都不会执行
private static void method1() {
Timer timer = new Timer();
TimerTask task1 = new TimerTask() {
@Override
public void run() {
log.debug("task 1");
//int i = 1 / 0;//任务一的出错会导致任务二无法执行
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
};
TimerTask task2 = new TimerTask() {
@Override
public void run() {
log.debug("task 2");
}
};
// 使用 timer 添加两个任务，希望它们都在 1s 后执行
// 但由于 timer 内只有一个线程来顺序执行队列中的任务，因此任务1的延时，影响了任务2的执行
timer.schedule(task1, 1000);//17:45:56 c.ThreadPool [Timer-0] - task 1
timer.schedule(task2, 1000);//17:45:58 c.ThreadPool [Timer-0] - task 2
}
5.2 Scheduled
任务调度线程池 ScheduledThreadPoolExecutor 继承 ThreadPoolExecutor：
- 使用内部类 ScheduledFutureTask 封装任务
- 使用内部类 DelayedWorkQueue 作为线程池队列
- 重写 onShutdown 方法去处理 shutdown 后的任务
- 提供 decorateTask 方法作为 ScheduledFutureTask 的修饰方法，以便开发者进行扩展
5.2.1 构造方法
构造方法：Executors.newScheduledThreadPool(int corePoolSize)
public ScheduledThreadPoolExecutor(int corePoolSize) {
// 最大线程数固定为 Integer.MAX_VALUE，保活时间 keepAliveTime 固定为 0
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
// 阻塞队列是 DelayedWorkQueue
new DelayedWorkQueue());
}
常用 API：
ScheduledFuture<?> schedule(Runnable/Callable<V>, long delay, TimeUnit u)：延迟执行任务ScheduledFuture<?> scheduleAtFixedRate(Runnable/Callable<V>, long initialDelay, long period, TimeUnit unit)：定时执行周期任务，不考虑执行的耗时，参数为初始延迟时间、间隔时间、单位ScheduledFuture<?> scheduleWithFixedDelay(Runnable/Callable<V>, long initialDelay, long delay, TimeUnit unit)：定时执行周期任务，考虑执行的耗时，参数为初始延迟时间、间隔时间、单位
5.2.2 schedule
延迟任务，但是出现异常并不会在控制台打印，也不会影响其他线程的执行，代码示例如下：
private static void method1() {
// 线程池大小为1时也是串行执行
ScheduledExecutorService executor = Executors.newScheduledThreadPool(2);
log.debug("begin");
// 添加两个任务，都在 1s 后同时执行
executor.schedule(() -> {
log.debug("任务1执行");
//int i = 1 / 0;
try { Thread.sleep(2000); } catch (InterruptedException e) { }
}, 1000, TimeUnit.MILLISECONDS);
executor.schedule(() -> {
log.debug("任务2执行");
}, 1000, TimeUnit.MILLISECONDS);
}
输出如下：
10:30:38.575 c.Test20 [main] - begin
10:30:39.619 c.Test20 [pool-1-thread-1] - 任务1执行
10:30:39.619 c.Test20 [pool-1-thread-2] - 任务2执行
5.2.3 scheduleAtFixedRate
定时任务 scheduleAtFixedRate，一次任务的启动到下一次任务的启动之间只要大于等于间隔时间，抢占到 CPU 就会立即执行
public static void method2() {
ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);
log.debug("start...");
pool.scheduleAtFixedRate(() -> {
log.debug("running...");
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}, 1, 1, TimeUnit.SECONDS);
}
输出如下：
10:34:21.377 c.Test20 [main] - start…
10:34:22.404 c.Test20 [pool-1-thread-1] - running…
10:34:25.405 c.Test20 [pool-1-thread-1] - running…
10:34:28.413 c.Test20 [pool-1-thread-1] - running…
10:34:31.414 c.Test20 [pool-1-thread-1] - running…
可以看到，初始延迟时间是1s，确实是开始后1s执行的，我们设置的是每一秒执行一次，但是每次任务都会执行3s，由于任务执行时间 > 间隔时间，间隔被『撑』到了 3s。
5.2.4 scheduleWithFixedDelay
一次任务的结束到下一次任务的启动之间等于间隔时间，抢占到 CPU 就会立即执行，这个方法才是真正的设置两个任务之间的间隔。
public static void method3(){
ScheduledExecutorService pool = Executors.newScheduledThreadPool(3);
log.debug("start...");
pool.scheduleWithFixedDelay(() -> {
log.debug("running...");
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}, 1, 1, TimeUnit.SECONDS);
}
输出如下：
10:43:33.071 c.Test20 [main] - start…
10:43:34.106 c.Test20 [pool-1-thread-1] - running…
10:43:38.132 c.Test20 [pool-1-thread-1] - running…
10:43:42.154 c.Test20 [pool-1-thread-2] - running…
10:43:46.170 c.Test20 [pool-1-thread-2] - running…
10:43:50.178 c.Test20 [pool-1-thread-3] - running…
每个任务执行3s，设置的每个任务之间的间隔为1s，于是任务执行之间的间隔为4s。
5.3 Tomcat线程池
Tomcat的执行流程如下：
- LimitLatch 用来限流，可以控制最大连接个数，类似 J.U.C 中de Semaphore
- Acceptor 只负责【接收新的 socket 连接】
- Poller 只负责监听 socket channel 是否有【可读的 I/O 事件】
- 一旦可读，封装一个任务对象（socketProcessor），提交给 Executor 线程池处理
- Executor 线程池中的工作线程最终负责【处理请求】
Tomcat 线程池扩展了 ThreadPoolExecutor，行为稍有不同
- 如果总线程数达到 maximumPoolSize，这时不会立刻抛 RejectedExecutionException 异常，而是再次尝试将任务放入队列，如果还失败，才抛出RejectedExecutionException 异常
该部分的在tomcat7.0.42中的源码如下：
public void execute(Runnable command, long timeout, TimeUnit unit) {
submittedCount.incrementAndGet();
try {
super.execute(command);
} catch (RejectedExecutionException rx) {
if (super.getQueue() instanceof TaskQueue) {
final TaskQueue queue = (TaskQueue)super.getQueue();
try {
if (!queue.force(command, timeout, unit)) {
submittedCount.decrementAndGet();
throw new RejectedExecutionException("Queue capacity is full.");
}
} catch (InterruptedException x) {
submittedCount.decrementAndGet();
Thread.interrupted();
throw new RejectedExecutionException(x);
}
} else {
submittedCount.decrementAndGet();
throw rx;
}
}
}
public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {
if ( parent.isShutdown() )
throw new RejectedExecutionException(
"Executor not running, can't force a command into the queue"
);
return super.offer(o,timeout,unit); //forces the item onto the queue, to be used if the task is rejected
}
6. Fork/Join
Fork/Join：线程池的实现，体现是分治思想，适用于能够进行任务拆分的 CPU 密集型运算，用于并行计算
任务拆分：将一个大任务拆分为算法上相同的小任务，直至不能拆分可以直接求解。跟递归相关的一些计算，如归并排序、斐波那契数列都可以用分治思想进行求解
-
Fork/Join 在分治的基础上加入了多线程，把每个任务的分解和合并交给不同的线程来完成，提升了运算效率
-
ForkJoin 使用 ForkJoinPool 来启动，是一个特殊的线程池，默认会创建与 CPU 核心数大小相同的线程池
-
任务有返回值继承 RecursiveTask，没有返回值继承 RecursiveAction
比如，我们实现用4个线程计算数字1到5的和，那么代码如下：
public static void main(String[] args) {
ForkJoinPool pool = new ForkJoinPool(4);
System.out.println(pool.invoke(new MyTask(5)));
//拆分 5 + MyTask(4) --> 4 + MyTask(3) -->
}
// 1~ n 之间整数的和
class MyTask extends RecursiveTask<Integer> {
private int n;
public MyTask(int n) {
this.n = n;
}
@Override
public String toString() {
return "MyTask{" + "n=" + n + '}';
}
@Override
protected Integer compute() {
// 如果 n 已经为 1，可以求得结果了
if (n == 1) {
return n;
}
// 将任务进行拆分(fork)
MyTask t1 = new MyTask(n - 1);
t1.fork();
// 合并(join)结果
int result = n + t1.join();
return result;
}
}
可以看到，compute里面的代码与递归的代码是一致的，但是，上面虽然使用了多线程，比如计算5+fork(4)的时候，其必须要等待for(4)的结果，这就造成了利用不充分，于是我们可以用分治改进：
class AddTask extends RecursiveTask<Integer> {
int begin;
int end;
public AddTask(int begin, int end) {
this.begin = begin;
this.end = end;
}
@Override
public String toString() {
return "{" + begin + "," + end + '}';
}
@Override
protected Integer compute() {
// 5, 5
if (begin == end) {
return begin;
}
// 4, 5 防止多余的拆分 提高效率
if (end - begin == 1) {
return end + begin;
}
// 1 5
int mid = (end + begin) / 2; // 3
AddTask t1 = new AddTask(begin, mid); // 1,3
t1.fork();
AddTask t2 = new AddTask(mid + 1, end); // 4,5
t2.fork();
int result = t1.join() + t2.join();
return result;
}
}
ForkJoinPool 实现了工作窃取算法来提高 CPU 的利用率：
- 每个线程都维护了一个双端队列，用来存储需要执行的任务
- 工作窃取算法允许空闲的线程从其它线程的双端队列中窃取一个任务来执行
- 窃取的必须是最晚的任务，避免和队列所属线程发生竞争，但是队列中只有一个任务时还是会发生竞争
