Java 8并行流如何处理抛出的异常？

Question

Java8并行流如何处理消费子句中抛出的异常，例如在forEach处理中？例如，以下代码：

final AtomicBoolean throwException = new AtomicBoolean(true);
IntStream.range(0, 1000)
    .parallel()
    .forEach(i -> {
        // Throw only on one of the threads.
        if (throwException.compareAndSet(true, false)) {
            throw new RuntimeException("One of the tasks threw an exception. Index: " + i);
        });

它是否立即停止已处理的元素？它是否等待已经启动的元素完成？它会等待所有的溪流完成吗？它是否在抛出异常后开始处理流元素？

它什么时候回来？就在例外之后？毕竟/部分元素是由使用者处理的？

在并行流抛出异常后，是否继续处理元素？(发现了发生这种情况的一个案例)。

这里有一般规则吗？

编辑 (15-11-2016)

为了确定并行流是否提前返回，我发现它不确定：

@Test
public void testParallelStreamWithException() {
    AtomicInteger overallCount = new AtomicInteger(0);
    AtomicInteger afterExceptionCount = new AtomicInteger(0);
    AtomicBoolean throwException = new AtomicBoolean(true);

    try {
        IntStream.range(0, 1000)
            .parallel()
            .forEach(i -> {
                overallCount.incrementAndGet();
                afterExceptionCount.incrementAndGet();
                try {
                    System.out.println(i + " Sleeping...");
                    Thread.sleep(1000);
                    System.out.println(i + " After Sleeping.");
                }
                catch (InterruptedException e) {
                    e.printStackTrace();
                }
                // Throw only on one of the threads and not on main thread.
                if (!Thread.currentThread().getName().equals("main") && throwException.compareAndSet(true, false)) {
                    System.out.println("Throwing exception - " + i);
                    throw new RuntimeException("One of the tasks threw an exception. Index: " + i);
                }
            });
        Assert.fail("Should not get here.");
    }
    catch (Exception e) {
        System.out.println("Cought Exception. Resetting the afterExceptionCount to zero - 0.");
        afterExceptionCount.set(0);
    }
    System.out.println("Overall count: " + overallCount.get());
    System.out.println("After exception count: " + afterExceptionCount.get());
}

延迟在抛出时返回，而不是从主线程抛出。这导致很多新元素在抛出异常后被处理。在我的机器上，大约有200个元素在抛出异常后被处理。但是，并不是所有的1000个元素都被处理了。这里的规矩是什么？为什么即使抛出异常也会处理更多的元素？

在删除not (!)符号时，会提前返回，从而导致在主线程中抛出异常。只有已经启动的元素完成了处理，而没有处理新的元素。这里的情况是早点回来。与之前的行为不一致。

我在这里错过了什么？

Answer 1

当异常在其中一个阶段被抛出时，它不会等待其他操作完成，该异常将被重新抛出给调用者。这就是ForkJoinPool处理这个问题的方式。

相反，例如，在并行运行时，只有在所有操作完成处理后(即使在需要完成所有操作之前就知道结果)，findFirst才会将结果呈现给调用方。

换句话说:它会提前返回，但所有的运行任务都要完成。

编辑以回答最后一个注释

Holger的回答(注释中的链接)解释了这一点，但这里有一些细节。

1)当杀死除主线程之外的所有线程时，您还将杀死这些线程应该处理的所有任务。所以这个数字实际上应该更多，因为有1000个任务和4个线程，我假设它返回3?：

int result = ForkJoinPool.getCommonPoolParallelism();

理论上有1000个任务，有4个线程，每个线程应该处理250个任务，然后杀死其中3个，这意味着750个任务丢失了。还有250个任务要执行，ForkJoinPool将跨越3个新线程来执行这250个剩余任务。

一些您可以尝试的事情，像这样改变您的流(使流不大小)：

IntStream.generate(random::nextInt).limit(1000).parallel().forEach

这一次，将会有更多的操作结束，因为初始的拆分索引是未知的，并且是由其他策略选择的。您还可以尝试更改以下内容：

 if (!Thread.currentThread().getName().equals("main") && throwException.compareAndSet(true, false)) {

对此：

 if (!Thread.currentThread().getName().equals("main")) {

这一次，除了main之外，您将始终关闭所有线程，直到某个时候，ForkJoinPool将不会创建任何新线程，因为任务太小，无法拆分，因此不需要其他线程。在这种情况下，完成的任务就更少了。

2)第二个示例，当您实际终止主线程时，正如代码的方式一样，您将不会看到其他线程的实际运行。改变它：

    } catch (Exception e) {
        System.out.println("Cought Exception. Resetting the afterExceptionCount to zero - 0.");
        afterExceptionCount.set(0);
    }

    // give some time for other threads to finish their work. You could play commenting and de-commenting this line to see a big difference in results. 
    TimeUnit.SECONDS.sleep(60);

    System.out.println("Overall count: " + overallCount.get());
    System.out.println("After exception count: " + afterExceptionCount.get());