对于斐波那契数的计算，我们都知道最容易理解的就是递归的方法：
斐波那契公式
public long recursiveFibonacci(int n) {
if (n < 2) {
return 1;
}
return recursiveFibonacci(n - 1) + recursiveFibonacci(n - 2);
}
当然这个递归也可以转化为迭代：
public long iterativeFibonacci(int n) {
long n1 = 1, n2 = 1;
long fi = 2; // n1 + n2
for (int i = 2; i <= n; i++) {
fi = n1 + n2;
n1 = n2;
n2 = fi;
}
return fi;
}
但是，对于以上两种方法，并不能并行化，因为后一项的值依赖于前一项，使得算法流程是串行的。所以引出了可以并行的计算斐波那契数的公式：
并行计算斐波那契数的公式
=>
并行计算斐波那契数的公式
f0 和 f1 都是 1 —— 很明显我们可以对
(1, 1; 1, 0) ^ n
首先我们定义一个 Matrix 类，用来表示一个 2*2 的矩阵：
public class Matrix {
/**
* 左上角的值
*/
public final BigInteger a;
/**
* 右上角的值
*/
public final BigInteger b;
/**
* 左下角的值
*/
public final BigInteger c;
/**
* 右下角的值
*/
public final BigInteger d;
public Matrix(int a, int b, int c, int d) {
this(BigInteger.valueOf(a), BigInteger.valueOf(b),
BigInteger.valueOf(c), BigInteger.valueOf(d));
}
public Matrix(BigInteger a, BigInteger b, BigInteger c, BigInteger d) {
this.a = a;
this.b = b;
this.c = c;
this.d = d;
}
/**
* multiply
*
* @param m multiplier
* @return
*/
public Matrix mul(Matrix m) {
return new Matrix(
a.multiply(m.a).add(b.multiply(m.c)), // a*a + b*c
a.multiply(m.b).add(b.multiply(m.d)), // a*b + b*d
c.multiply(m.a).add(d.multiply(m.c)), // c*a + d*c
c.multiply(m.b).add(d.multiply(m.d)));// c*b + d*d
}
/**
* power of exponent
*
* @param exponent
* @return
*/
public Matrix pow(int exponent) {
Matrix matrix = this.copy();
for (int i = 1; i < exponent; i++) {
matrix = matrix.mul(this);
}
return matrix;
}
public Matrix copy() {
return new Matrix(a, b, c, d);
}
}
然后我们来比较迭代和并行的效率：
我们先设置并行使用的线程数为 1，即单线程。
public static void main(String[] args) throws Exception {
final int ITEM_NUM = 500000; // 计算斐波那契数列的第 ITEM_NUM 项
System.out.println("开始迭代计算...");
long begin = System.nanoTime();
BigInteger fi1 = iterativeFibonacci(ITEM_NUM);
long end = System.nanoTime();
double time = (end - begin) / 1E9;
System.out.printf("迭代计算用时: %.3f\n\n", time);
/* ------------------------------ */
System.out.println("开始并行计算...");
begin = System.nanoTime();
BigInteger fi2 = parallelFibonacci(ITEM_NUM, 1);
end = System.nanoTime();
time = (end - begin) / 1E9;
System.out.printf("并行计算用时: %.3f\n\n", time);
System.out.println("fi1 == fi2：" + (fi1.equals(fi2)));
}
static BigInteger iterativeFibonacci(int n) {
BigInteger n1 = BigInteger.ONE;
BigInteger n2 = BigInteger.ONE;
BigInteger fi = BigInteger.valueOf(2); // n1 + n2
for (int i = 2; i <= n; i++) {
fi = n1.add(n2);
n1 = n2;
n2 = fi;
}
return fi;
}
static BigInteger parallelFibonacci(int itemNum, int threadNum) throws Exception {
final Matrix matrix = new Matrix(1, 1, 1, 0);
final Matrix primary = new Matrix(1, 0, 1, 0); // (f0, 0; f1, 0)
final int workload = itemNum / threadNum; // 每个线程要计算的 相乘的项数
// (num / threadNum) 可能存在除不尽的情况，所以最后一个任务计算所有剩下的项数
final int lastWorkload = itemNum - workload * (threadNum - 1);
List<Callable<Matrix>> tasks = new ArrayList<>(threadNum);
for (int i = 0; i < threadNum; i++) {
if (i < threadNum - 1) {
// 为了简洁，使用 Lambda 表达式替代要实现 Callable<Matrix> 的匿名内部类
tasks.add(() -> matrix.pow(workload));
} else {
tasks.add(() -> matrix.pow(lastWorkload));
}
}
ExecutorService threadPool = Executors.newFixedThreadPool(threadNum);
List<Future<Matrix>> futures = threadPool.invokeAll(tasks); // 执行所有任务，invokeAll 会阻塞直到所有任务执行完毕
Matrix result = primary.copy();
for (Future<Matrix> future : futures) { // (matrix ^ n) * (f0, 0; f1, 0)
result = result.mul(future.get());
}
threadPool.shutdown();
return result.c;
}
并行的线程数为 1 的结果
可以看到单线程情况下，使用矩阵运算的效率大概只有迭代计算的 1/3 左右 —— 既然如此，那我们耍流氓的把并行的线程数改为 10 线程吧：
BigInteger fi2 = parallelFibonacci(ITEM_NUM, 10); // 10 线程并行计算
并行的线程数为 10 的结果
此时并行计算的用时碾压了迭代计算 —— 迭代计算委屈的哭了，并行计算这流氓耍的相当漂亮。
好像有点不对劲，我这篇文章的标题似乎是 使用并行流 —— 并行流呢？
其实前面都是铺垫 :) 在 parallelFibonacci 方法中，我们使用了线程池来并行的执行任务，我们来尝试将 parallelFibonacci 改为流式（即基于 Stream）风格的代码：
static BigInteger streamFibonacci(int itemNum, int threadNum) {
final Matrix matrix = new Matrix(1, 1, 1, 0);
final Matrix primary = new Matrix(1, 0, 1, 0);
final int workload = itemNum / threadNum;
final int lastWorkload = itemNum - workload * (threadNum - 1);
// 流式API
return IntStream.range(0, threadNum) // 产生 [0, threadNum) 区间，用于将任务切分
.parallel() // 使流并行化
.map(i -> i < threadNum - 1 ? workload : lastWorkload)
.mapToObj(w -> matrix.pow(w)) // map -> mN = matrix ^ workload
.reduce((m1, m2) -> m1.mul(m2)) // reduce -> m = m1 * m2 * ... * mN
.map(m -> m.mul(primary)) // map -> m = m * primary
.get().c; // get -> m.c
}
依旧在 10 线程的环境下运行下看看：
public static void main(String[] args) throws Exception {
...
/* ------------------------------ */
System.out.println("开始流式并行计算...");
begin = System.nanoTime();
BigInteger fi3 = streamFibonacci(ITEM_NUM, 10);
end = System.nanoTime();
time = (end - begin) / 1E9;
System.out.printf("流式并行计算用时: %.3f\n\n", time);
System.out.println("fi1 == fi2：" + (fi1.equals(fi2)));
System.out.println("fi1 == fi3：" + (fi1.equals(fi3)));
}
10 线程并行计算的结果
是的，使用并行流就是这么的简单，只要你会使用 Stream API —— 给它加上 .parallel() —— 它就并行化了。写了这么多年的 Java 代码，从 Java6 到 Java7 再到 Java8，这一刻，我真的感动了（容我擦擦眼泪）。
而且我们可以看到，在线程数相同的情况下，使用 streamFibonacci（并行流）时，用时要比parallelFibonacci 方法更短。为了验证，我夸张一点，将线程数提高到 32：
BigInteger fi2 = parallelFibonacci(ITEM_NUM, 32);
...
BigInteger fi3 = streamFibonacci(ITEM_NUM, 32);
32 线程并行计算的结果
可以看到，此时 parallelFibonacci 的运行时间反而比 10 线程的时候更长了，而 streamFibonacci 使用的时间却更短了 —— 流式 API 厉害了！
但这是什么原因呢？这个问题留给有兴趣的读者思考和探究吧。
值得注意的是，并行流的底层实现是基于 ForkJoinPool 的，并且使用的是一个共享的 ForkJoinPool —— ForkJoinPool.commonPool()。为了充分利用处理器资源和提升程序性能，我们应该尽量使用并行流来执行 CPU 密集的任务，而不是 IO 密集的任务 —— 因为共享池中的线程数量是有限的，如果共享池中某些线程执行 IO 密集的任务，那么这些线程将长时间处于等待 IO 操作完成的状态，一旦共享池中的线程耗尽，那么程序中其他想继续使用并行流的地方就需要等待，直到有空闲的线程可用，这会在很大程度上影响到程序的性能。所以使用并行流之前，我们要注意到这个细节。
