stream与parallelStream

Question

我有这样的测试代码：

List<Integer> list = new ArrayList<>(1000000);

for(int i=0;i<1000000;i++){
    list.add(i);
}

List<String> values = new ArrayList<>(1000000);

list.stream().forEach(
    i->values.add(new Date().toString())
);

System.out.println(values.size()); 

运行这个程序，我得到了一个正确的输出: 1000000。

但是，如果我将stream()更改为parallelStream()，如下所示：

 list.parallelStream().forEach(
    i->values.add(new Date().toString())
 );

我得到了一个随机输出，例如: 920821。

怎么了？

Answer 1

ArrayList不同步。试图同时向其添加元素是没有定义的。来自forEach

对于并行流管道，此操作并不保证尊重流的遭遇顺序，因为这样做将牺牲并行性的好处。对于任何给定的元素，操作可以在任何时间执行，也可以在库选择的线程中执行。

在第二个示例中，最终会有多个线程同时调用数组列表上的add，而ArrayList文档指出：

注意，此实现不是同步的。如果多个线程同时访问ArrayList实例，并且至少有一个线程在结构上修改列表，则必须在外部同步它。

错解

如果将ArrayList的用法更改为Vector，则会得到正确的结果，因为这个列表实现是同步的。它的Javadoc说：

与新的集合实现不同，Vector是同步的。

然而，不要用它！更重要的是，由于显式同步，它最终可能会变慢。

正确的方法

为了避免这种情况，Stream通过使用可变约简方法提供了collect范例。以下是

List<String> values = list.stream().map(i -> "foo").collect(Collectors.toList());

将始终提供正确的结果，无论是否并行运行。Stream管道内部处理并发性和确保在并行流的收集操作中使用非并发收集器是安全的。。Collectors.toList()是一个内置的收集器，它将流的元素积累到列表中。

Answer 2

使用一个消费者，你必须担心线程的安全。一种更简单的解决方案--让Stream积累结果。

List<String> values = IntStream.range(0, 1_000_000).parallel()
                               .mapToObj(i -> new Date().toString())
                               .collect(Collectors.toList());

避免使用向量这样的线程安全收集器的一个关键原因是，它要求每个线程获得一个共享锁，这是一个瓶颈，即您将花费时间获取和释放锁，每次只能有一个线程可以访问它。您可以很容易地得到比单独使用一个线程更慢的解决方案。

Answer 3

values.add(String)并不是线程安全的。当您在没有同步的情况下从不同的线程调用此方法时，并不能保证它将按预期的方式工作。

要解决这个问题，您可以：

• 使用线程安全集合，如Vector或CopyOnWriteArrayLis。
• 显式同步代码。例如，将synchronize(this){values.add(new Date().toString())}放入代码中。注i->在同步块外部
• 或者在本例中，映射一些元素来获得新的流，比如@PeterLawrey，答案是：IntStream.range(0, 1_000_000).parallel().mapToObj(i -> new Date().toString()).collect(Collectors.toList());