Java8多线程ForkJoinPool：处理异常

引入Java8 lambda背后的主要动机之一是能够尽可能轻松地使用多核的能力（请参阅精通Lambdas：多核世界中的Java编程）。只需将代码从collection.stream()...更改为collection.parallelStream()...您就可以使用即时多线程，从而为您的计算机带来所有CPU功能。（在这一点上，让我们忽略争用。）

如果打印出parallelStream所使用的线程的名称，您会注意到它们与ForkJoin框架所使用的线程相同，如下所示：

[ForkJoinPool.commonPool-worker-1]
[ForkJoinPool.commonPool-worker-2]

请参阅本杰明·温特伯格的博客，以获取一个很好的示例。

现在，在Java 8中，您可以将这个commonPool与ForkJoinPool commonPool（）上的新方法直接一起使用。这将返回带有commonPool线程的ForkJoinPool（这是一个ExecutorService）实例–与parallelStream中使用的线程相同。这意味着您直接使用commonPool进行的任何工作都可以与parallelStream中完成的工作很好地配合，尤其是线程调度和线程之间的窃取。

让我们来看一个如何使用ForkJoin的示例，尤其是在处理棘手的异常主题时。

首先，通过调用ForkJoin.commonPool()获得commonPool的实例。您可以使用submit()方法向其提交任务。因为我们使用的是Java8，所以我们可以传入lambda表达式，这确实很不错。与所有ExecutorService实现一样，您可以将Runnable或Callable实例传递给Runnable submit() 。当您将lambda传递给Submit方法时，它将通过检查方法签名将其自动转换为Runnable或Callable 。

这导致一个有趣的问题，突出了lambda如何工作。假设您有一个返回类型为void的方法（如Runnable），但抛出了一个已检查的异常（如Callable）。参见方法throwException()
在下面的代码清单中可以找到这样的例子。如果您编写此代码，它将无法编译。

Future task1 = commonPool.submit(() -> {throwException("task 1");});

这样做的原因是，由于void返回类型，编译器假定您正在尝试创建Runnable。当然，Runnable不能抛出异常。要解决此问题，您需要强制编译器了解您正在创建一个Callable，使用此代码技巧可以允许抛出Exception。

Future task1 = commonPool.submit(() -> {throwException("task 1");return null;});

这有点混乱，但可以完成工作。可以说，编译器本来可以解决这个问题。

下面的完整代码清单中还有两点要强调。第一，您可以使用commonPool.getParallelism()看到池中有多少个线程可用。可以使用参数'-Djava.util.concurrent.ForkJoinPool.common.parallelism'进行调整。第二，注意如何解开ExecutionException，以便您的代码仅向其调用者显示IOException，而不是非特定的ExecutionException。另请注意，此代码在第一个异常时失败。如果要收集所有异常，则必须适当地构造代码，可能会返回异常列表。或者更巧妙地抛出一个包含基础异常列表的自定义异常。

public class ForkJoinTest {public void run() throws IOException{ForkJoinPool commonPool = ForkJoinPool.commonPool();Future task1 = commonPool.submit(() -> {throwException("task 1");return null;});Future task2 = commonPool.submit(() -> {throwException("task 2");return null;});System.out.println("Do something while tasks being " +"executed on " + commonPool.getParallelism()+ " threads");try {//wait on the result from task2task2.get();//wait on the result from task1task1.get();} catch (InterruptedException e) {throw new AssertionError(e);} catch (ExecutionException e) {Throwable innerException = e.getCause();if (innerException instanceof RuntimeException) {innerException = innerException.getCause();if(innerException instanceof IOException){throw (IOException) innerException;}}throw new AssertionError(e);}}public void throwException(String message) throws IOException,InterruptedException {Thread.sleep(100);System.out.println(Thread.currentThread() + " throwing IOException");throw new IOException("Throw exception for " + message);}public static void main(String[] args) throws IOException{new ForkJoinTest().run();}
}