线程池和Fork/Join的最终目标是相似的:都希望充分利用可用的CPU资源,以获得最大吞吐量。最大吞吐量意味着应该在长时间内完成尽可能多的任务。要做到这一点,至少需要运行与可用CPU数量相同的线程,因为运行较少的线程会导致核心未被使用。最多应有与可用CPU数量相同的线程在运行,因为运行更多的线程将为调度程序创建额外的负载,调度程序将CPU分配给不同的线程,这会导致一些CPU时间用于调度程序而不是计算任务。
因此,我们确定了要实现最大吞吐量,需要具有与CPU数量完全相同的线程数。在Oracle的示例中,您可以使用一个具有线程数等于可用CPU数量的固定大小线程池或使用线程池。这不会有任何区别,你是对的!
那么什么情况下会遇到线程池的麻烦呢?就是当线程被阻塞时,因为您的线程正在等待另一个任务完成。假设以下示例:
class AbcAlgorithm implements Runnable {
public void run() {
Future<StepAResult> aFuture = threadPool.submit(new ATask());
StepBResult bResult = stepB();
StepAResult aResult = aFuture.get();
stepC(aResult, bResult);
}
}
这里展示的算法由三个步骤 A、B 和 C 组成。A 和 B 可以相互独立执行,但是步骤 C 需要依赖于步骤 A 和 B 的结果。这个算法提交任务 A 到线程池中并直接执行任务 B。之后,该线程将等待任务 A 完成,并继续执行步骤 C。如果 A 和 B 同时完成,则一切正常。但是如果 A 比 B 花费更长的时间呢?这可能是因为任务 A 的特性所致,也可能是因为一开始没有为任务 A 准备好线程,需要等待。 (如果只有一颗 CPU 可用,因此您的线程池只有一个线程,这甚至会导致死锁,但现在这不是重点)。关键是刚刚执行任务 B 的线程会阻塞整个线程。由于我们有与 CPU 数相同的线程数,并且其中一个线程被阻塞了,这意味着
一个 CPU 空闲。
Fork/Join 可以解决这个问题:在 Fork/Join 框架中,可以按照以下方式编写相同的算法:
class AbcAlgorithm implements Runnable {
public void run() {
ATask aTask = new ATask());
aTask.fork();
StepBResult bResult = stepB();
StepAResult aResult = aTask.join();
stepC(aResult, bResult);
}
}
看起来一样,是吗?然而,关键在于aTask.join
不会阻塞。这里就涉及到工作窃取的问题:线程将寻找过去被分叉的其他任务并继续执行它们。首先,它会检查自己分叉的任务是否已经开始处理。如果A还没有被另一个线程启动,那么它将接着做A,否则它将检查其他线程的队列并窃取他们的工作。一旦另一个线程的任务完成了,它将检查A是否已经完成。如果是,则上述算法可以调用stepC
。否则,它将寻找另一个要窃取的任务。因此,fork/join池甚至在面对阻塞操作时也能实现100%的CPU利用率。
但是有一个陷阱:工作窃取仅适用于ForkJoinTask
的join
调用。无法针对等待另一个线程或等待I/O操作等外部阻塞操作完成的情况进行操作。那么等待I/O完成是常见的任务怎么办?在这种情况下,如果我们可以向Fork/Join池添加一个额外的线程,该线程将在阻塞操作完成后立即停止,这将是第二好的方法。如果我们使用ManagedBlocker
,那么ForkJoinPool
实际上可以做到这一点。
Fibonacci
《JavaDoc for RecursiveTask》中有一个使用Fork/Join计算斐波那契数列的例子。对于经典的递归解决方案,请参见:
public static int fib(int n) {
if (n <= 1) {
return n;
}
return fib(n - 1) + fib(n - 2);
}
根据JavaDocs的解释,这是一个计算斐波那契数列的相当低效的方法,因为该算法的复杂度为O(2^n),虽然有更简单的方法可用。但是该算法非常简单易懂,因此我们坚持使用它。假设我们想要使用Fork/Join来加快速度。一个天真的实现看起来像这样:
class Fibonacci extends RecursiveTask<Long> {
private final long n;
Fibonacci(long n) {
this.n = n;
}
public Long compute() {
if (n <= 1) {
return n;
}
Fibonacci f1 = new Fibonacci(n - 1);
f1.fork();
Fibonacci f2 = new Fibonacci(n - 2);
return f2.compute() + f1.join();
}
}
此任务分为的步骤太短,因此性能表现非常差,但您可以看到框架通常运作得非常好:两个加数可以独立计算,但是我们需要它们来构建最终结果。因此,一半在另一个线程中完成。尝试使用线程池执行相同操作时,请注意避免死锁(可能,但不如简单)。
仅为完整起见:如果您确实想使用此递归方法计算斐波那契数列,这里有一个经过优化的版本:
class FibonacciBigSubtasks extends RecursiveTask<Long> {
private final long n;
FibonacciBigSubtasks(long n) {
this.n = n;
}
public Long compute() {
return fib(n);
}
private long fib(long n) {
if (n <= 1) {
return 1;
}
if (n > 10 && getSurplusQueuedTaskCount() < 2) {
final FibonacciBigSubtasks f1 = new FibonacciBigSubtasks(n - 1);
final FibonacciBigSubtasks f2 = new FibonacciBigSubtasks(n - 2);
f1.fork();
return f2.compute() + f1.join();
} else {
return fib(n - 1) + fib(n - 2);
}
}
}
这种方法能够保持子任务更小,因为只有当n > 10 && getSurplusQueuedTaskCount() < 2
条件成立时才进行拆分,这意味着有超过100个方法调用要执行(n > 10
),并且没有很多任务在等待执行(getSurplusQueuedTaskCount() < 2
)。
在我的电脑上(4核(8核超线程),Intel(R) Core(TM) i7-2720QM CPU @ 2.20GHz),使用经典方法计算fib(50)
需要64秒,而使用Fork/Join方法只需要18秒,这是相当明显的性能提升,虽然不如理论上可能的那么多。
总结
- 在您的示例中,Fork/Join与经典线程池相比没有优势。
- 当涉及阻塞时,Fork/Join可以大大提高性能。
- Fork/Join规避了一些死锁问题。