字段上的 forEach 内部的 Java 8 Streams 干扰

Question

考虑以下使用 java 8 流的愚蠢程序：

private int biggestInt;

private void run() {
    ExecutorService executor = Executors.newWorkStealingPool();

    List<Callable<Integer>> callables = new ArrayList<>();

    for (int i = 0; i<50; i++) {
        callables.add(randomInt());
    }

    try {
        executor.invokeAll(callables)
            .stream()
            .map(future -> {
                    try {
                        return future.get();
                    } catch (Exception e) {
                        throw new IllegalStateException(e);
                    }
                })
            .forEach(this::compareBiggestInt);
    } catch (InterruptedException e) { /* do nothing */ }
}

private Callable<Integer> randomInt() {
    return () -> {
        Random random = new Random(System.currentTimeMillis());
        return random.nextInt();
    };
}

private void compareBiggestInt(Integer in) {
    if (in > biggestInt)
        biggestInt = in;
}

我的问题是，forEach(this::compareBiggestInt) 是否并行执行，因此会在最大 Int 上引入竞争条件？

如果是这样，我怎样才能避免这种竞争条件？ 例如，我可以像下面这样更改方法吗？

private synchronized void compareBiggestInt(Integer in) {[...]}

感谢任何帮助！

Answer 1

不，forEach 不是并行执行的。这将打破forEach 在与stream() 而非parallelStream() 一起使用时的预期行为的一般约定，并且不受您引入ExecutorService 这一事实的影响。

invokeAll() 实际上返回了Future 实例中的List，这些实例已完成或超时。因此，在您与流交互时，并行部分已经完成。

Answer 2

这里有一些问题。第一：

return () -> {
    Random random = new Random(System.currentTimeMillis());
    return random.nextInt();
};

执行速度可能如此之快（我可以轻松重现），这将始终返回相同的值。

我建议你至少删除 millis：

private static Callable<Integer> randomInt() {
    return () -> {
        Random random = new Random();
        int x = random.nextInt(100);
        System.out.println(x);
        return x;
    };
}

或者更好地使用ThreadLocalRandom.current().nextInt(100)

我还更改了nextInt 以返回到[0.. 100] 的范围内，因为nextInt 可以返回一个负值并假设您返回50 个负值，然后您的最大值将是zero（默认值） biggestInt;这显然是错误的。

然后你的流是sequential 并且在每个map 操作中你阻塞直到Future.get 完成。因此，您的 forEach 由单个线程执行。

Answer 3

forEach 不在并行 流中执行。实际执行异步任务的是executor。 Stream#map 操作将等到所有 Futures 完成。

如果你想要一个操作在并行流中执行，你应该使用reduction operation:Stream#reduce。例如：

biggestInt = executor.invokeAll(callables)
        .parallelStream()
        .map(...)// same with yours
        .reduce(BinaryOperator.maxBy(Comparator.naturalOrder()))
        .orElse(null);

Answer 4

您不使用parallel 流，因此您的流是顺序的。如果您想确保您的流按顺序完成，请将 .sequential() 方法添加到您的流中。

来自docs：

default Stream<E> stream()  
Returns a sequential Stream with this collection as its source.

Answer 5

假设您正在并行运行流（我将代码更改为使用“parallelStream”），您必须保护对共享可变变量的所有更改。

例如，在下面的代码中，我在方法“compareBiggestInt”中使用“同步”来保护对变量“biggestInt”的所有访问。 （如果您删除“同步”并运行以下代码，您可以看到方法“compareBiggestInt”中确实存在竞争条件）

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


public class ParallelStreamExample  {

    private volatile int biggestInt;

    public static void main(String[] args) {
        ParallelStreamExample parallelStreamExample = new ParallelStreamExample();
        parallelStreamExample.doTheWork();
    }



    private void doTheWork() {
        ExecutorService executor = Executors.newWorkStealingPool();

        List<Callable<Integer>> callables = new ArrayList<>();

        for (int i = 0; i < 5; i++) {
            callables.add(randomInt());
        }

        try {
            executor.invokeAll(callables)
                    .parallelStream()
                    .map(future -> {
                        try {
                            return future.get();
                        } catch (Exception e) {
                            throw new IllegalStateException(e);
                        }
                    })
                    .forEach(this::compareBiggestInt);
        } catch (InterruptedException e) { /* do nothing */ }
    }

    private Callable<Integer> randomInt() {
        return () -> {
            Random random = new Random();
            return random.nextInt(10);
        };
    }

    private synchronized void compareBiggestInt(Integer in)  {
        System.out.println("in:" + in + " - current biggestint = " + biggestInt);
        if (in > biggestInt) {
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            biggestInt = in;
        }
        System.out.println("in:" + in + " - current biggestint = " + biggestInt);
    }
}