Java 8并行流如何在抛出的异常中表现？ [英] How do Java 8 parallel streams behave on a thrown exception?

问题描述

Java 8并行流如何在consume子句中的抛出异常上运行，例如在 forEach 处理？例如，以下代码：

How do Java 8 parallel streams behave on a thrown exception in the consuming clause, for example in forEach handling? For example, the following code:

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);
        });

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

Does it stop the handled elements immediately? Does it wait for the already started elements to finish? Does it wait for all the stream to finish? Does it start handling stream elements after the exception is thrown?

什么时候返回？异常后立即？在所有/部分元素由消费者处理之后？

When does it return? Immediately after the exception? After all/part of the elements were handled by the consumer?

在并行流引发异常后，是否继续处理元素？ （发现这种情况发生的情况）。

Do elements continue being handled after the parallel stream threw the exception? (Found a case where this happened).

这里有一般规则吗？

编辑（2016年11月15日）

EDIT (15-11-2016)

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

Trying to determine if the parallel stream returns early, I found that it's not determinate:

@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());
}

延迟退货当不从主线程投掷时。这导致在抛出异常后处理大量 new 元素。在我的机器上，抛出异常后处理了大约200个元素。但是，并非所有1000个元素都得到了处理。那么这里的规则是什么？为什么即使抛出异常也会处理更多元素？

Late return when throwing not from the main thread. This caused a lot of new elements to be handled way after the exception was thrown. On my machine, about 200 elements were handled after the exception was thrown. BUT, not all 1000 elements were handled. So what's the rule here? Why more elements were handled even though the exception was thrown?

提前返回时删除not（！）签名，导致在主线程中抛出异常。只有已经启动的元素已完成处理，并且没有处理新的元素。这里的情况很早就回来了。与以前的行为不一致。

Early return when removing the not (!) sign, causing the exception to be thrown in the main thread. Only the already started elements finished processing and no new ones were handled. Returning early was the case here. Not consistent with the previous behavior.

我在这里缺少什么？

推荐答案

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

When an exception is thrown in one of the stages, it does not wait for other operations to finish, the exception is re-thrown to the caller. That is how ForkJoinPool handles that.

相比之下，findFirst例如并行运行时，只有在所有操作完成处理后才会将结果呈现给调用者（即使结果在需要之前已知）完成所有操作）。

In contrast findFirst for example when run in parallel, will present the result to the caller only after ALL operations have finished processing (even if the result is known before the need to finish of all operations).

换句话说：它会提前返回，但会完成所有正在运行的任务。

Put in other words : it will return early, but will leave all the running tasks to finish.

编辑回答最后的评论

Holger's解释了这一点回答（链接在评论中），但这里有一些细节。

This is very much explained by Holger's answer (link in comments), but here are some details.

1）当杀死所有BUT主线程时，你也杀死了应该由这些线程处理的所有任务。所以这个数字应该实际上大约250，因为有1000个任务和4个线程，我认为这会返回3？：

1) When killing all BUT the main thread, you are also killing all the tasks that were supposed to be handled by these threads. So that number should actually be more around 250 as there are 1000 tasks and 4 Threads, I assume this returns 3?:

int result = ForkJoinPool.getCommonPoolParallelism();

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

Theoretically there are 1000 tasks, there are 4 threads, each supposed to handle 250 tasks, then you kill 3 of them meaning 750 tasks are lost. There are 250 tasks left to execute, and ForkJoinPool will span 3 new threads to execute these 250 left tasks.

你可以尝试一些事情，改变你的像这样的流（使流不大）：

A few things you can try, change your stream like this (making the stream not sized):

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

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

This time, there would be many more operations ending, because the initial split index is unknown and chosen by some other strategy. What you could also try is change this :

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

到此：

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

这次你总是会杀死除main之外的所有线程，直到某个点，ForkJoinPool不会创建任何新线程作为任务太小而不能拆分，因此不需要其他线程。在这种情况下，更少的任务将完成。

This time you would always kill all threads besides main, until a certain point, where no new threads will be created by ForkJoinPool as the task is too small to split, thus no need for other threads. In this case even less tasks would finish.

2）你的第二个例子，当你真正杀死主线程时，就像代码的方式一样，你不会看到实际的运行其他线程。更改它：

2) Your second example, when you actually kill the main thread, as the way code is, you will not see the actual running of other threads. Change it :

    } 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());

这篇关于Java 8并行流如何在抛出的异常中表现？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！