JDK 7 中的 Fork/Join 模式

发布: 2008-6-26 11:04 | 作者: 不详 | 来源: 测试时代编辑整理 | 查看: 89次 | 进入软件测试论坛讨论

    private void sequentiallySort(long[] array, int lo, int hi) {
        Arrays.sort(array, lo, hi + 1);
    }
}

在清单 1 中，SortTask 首先通过 partition() 方法将数组分成两个部分。随后，两个子任务将被生成并分别排序数组的两个部分。当子任务足够小时，再将其分割为更小的任务反而引起性能的降低。因此，这里我们使用一个 THRESHOLD，限定在子任务规模较小时，使用直接排序，而不是再将其分割成为更小的任务。其中，我们用到了 RecursiveAction 提供的方法 coInvoke()。它表示：启动所有的任务，并在所有任务都正常结束后返回。如果其中一个任务出现异常，则其它所有的任务都取消。coInvoke() 的参数还可以是任务的数组。

现在剩下的工作就是将 SortTask 提交到 ForkJoinPool 了。ForkJoinPool() 默认建立具有与 CPU 可使用线程数相等线程个数的线程池。我们在一个 JUnit 的 test 方法中将 SortTask 提交给一个新建的 ForkJoinPool：

清单 2. 新建的 ForkJoinPool
@Test
public void testSort() throws Exception {
    ForkJoinTask sort = new SortTask(array);
    ForkJoinPool fjpool = new ForkJoinPool();
    fjpool.submit(sort);
    fjpool.shutdown();

fjpool.awaitTermination(30, TimeUnit.SECONDS);

assertTrue(checkSorted(array));
}

在上面的代码中，我们用到了 ForkJoinPool 提供的如下函数：

submit()：将 ForkJoinTask 类的对象提交给 ForkJoinPool，ForkJoinPool 将立刻开始执行 ForkJoinTask。
shutdown()：执行此方法之后，ForkJoinPool 不再接受新的任务，但是已经提交的任务可以继续执行。如果希望立刻停止所有的任务，可以尝试 shutdownNow() 方法。
awaitTermination()：阻塞当前线程直到 ForkJoinPool 中所有的任务都执行结束。
并行快速排序的完整代码如下所示：

清单 3. 并行快速排序的完整代码
package tests;

import static org.junit.Assert.*;

import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.TimeUnit;