Java 8 中并行流的使用注意事项

本贴最后更新于 1780 天前,其中的信息可能已经沧海桑田

Java 8 中并行流的使用注意事项

使用 Java 8 中的并行流已经有很长一段时间,有很多时候对于集合的处理可以不再去使用多线程,而直接使用并行流,因为并行流是线程安全的。但是我发现有很多人也包括我在使用并行流的时候,并没有正确的去使用它,而导致一些问题,甚至怀疑并行流到底是不是线程安全的。

为了找到避免这个不正确用法坑,为此我写了一些并行流的示例。

错误的使用方式

这里我写了一个很常见的错误用法:

List<Integer> integers = new ArrayList<>();
for (int i = 0; i < 1000; i++ ) {
    integers.add(i);
}

List<String> strings = new ArrayList<>();
integers.parallelStream().forEach(i -> strings.add(i.toString()));

System.out.println(strings);

输出结果:

Exception in thread "main" java.lang.ArrayIndexOutOfBoundsException
	at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
	...
Caused by: java.lang.ArrayIndexOutOfBoundsException: 823
	at java.util.ArrayList.add(ArrayList.java:463)
	at xin.codedream.parallel.ParallelStreamExample.lambda$main$0(ParallelStreamExample.java:18)
	...

是的,这种错误的使用方式会导致 下标越界 或者 空指针异常。这个时候就会有人怀疑并行流的线程安全问题。

把 ArrayList 换成线程安全的 CopyOnWriteArrayList 或者使用 Collections.synchronizedList() 来创建一个线程安全的集合。

使用线程安全的集合

这是使用 CopyOnWriteArrayList 的示例:

List<Integer> integers = new ArrayList<>();
for (int i = 0; i < 1000; i++ ) {
    integers.add(i);
}

List<String> strings = new CopyOnWriteArrayList<>();
integers.parallelStream().forEach(i -> strings.add(i.toString()));

System.out.println(strings);

输出结果:

[656, 657, 658...873, 874]

值得庆幸的是这段代码并没有报错。但,我们去看看它的代码你会发现使用了 :

java.util.concurrent.CopyOnWriteArrayList#add(E)
...
public boolean add(E e) {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        Object[] elements = getArray();
        int len = elements.length;
        Object[] newElements = Arrays.copyOf(elements, len + 1);
        newElements[len] = e;
        setArray(newElements);
        return true;
    } finally {
        lock.unlock();
    }
}

同样使用 Collections.synchronizedList(),也能创建线程安全的集合,同样的我们去看看它的代码:

java.util.Collections.SynchronizedList#add
...
public void add(int index, E element) {
    synchronized (mutex) {list.add(index, element);}
}

同样的,也是用锁。线程安全的集合的性能有一定下降,那我们来看看正确的使用方法。

并行流正确的使用方法

List<Integer> integers = new ArrayList<>();
for (int i = 0; i < 1000; i++) {
    integers.add(i);
}

List<String> strings = integers.parallelStream().collect(ArrayList::new, (arrayList, i) -> {
    arrayList.add(i.toString());
}, List::addAll);

System.out.println(strings);

输出结果:

[0, 1, 2...998, 999]

是的,这就是并行流正确的使用方法,而且可以看到还是数字也是连续的,并不像使用线程安全的集合那样并非从头到尾都是连续的数字。

排查不正确使用方式出错的原因

为了找出这个不正确使用方式导致的出错的原因,我看了看相关的源代码 java.util.stream.ForEachOps.ForEachTask#compute

compute

中的这一行代码 task.propagateCompletion();

task.propagateCompletion()

在不正确的方式下执行了多次,而正确方式下并行流,却没有执行过一次。

我猜测,导致使用出错的原因是因为,并行流在分割任务时由于之前分割的任务还没执行完,而已经执行任务的线程,去将未执行完或者执行到一半的线程的任务再次进行分割后去执行,而使用的是非线程安全的类被其他线程重复的执行,而导致出错或者结果不对。

这次排查,我并没有很找出导致出错的真正原因,所以只是猜测它可能是因为这个原因而导致出错。真正排查出这个原因后,将会更新此文。

blog:https://www.codedream.xin

  • Java

    Java 是一种可以撰写跨平台应用软件的面向对象的程序设计语言,是由 Sun Microsystems 公司于 1995 年 5 月推出的。Java 技术具有卓越的通用性、高效性、平台移植性和安全性。

    3187 引用 • 8213 回帖

相关帖子

欢迎来到这里!

我们正在构建一个小众社区,大家在这里相互信任,以平等 • 自由 • 奔放的价值观进行分享交流。最终,希望大家能够找到与自己志同道合的伙伴,共同成长。

注册 关于
请输入回帖内容 ...
  • Not-Found
    作者

    不一定会报错,而是有可能会报错。因为 fork/join 的过程中,都是“正确”的处理完了。

  • 其他回帖
  • cloudlang
    List<Integer> integers = new ArrayList<>(); 
    for (int i = 0; i < 1000; i++ ) { integers.add(i); } 
    List<String> strings = new ArrayList<>(); 
    integers.parallelStream().forEach(i -> strings.add(i.toString()));
     System.out.println(strings);  
    

    写了个测试方法,发现这个代码块报错不是 100%,原因没时间深究,先不地道的挖坑。

    2 回复
  • Not-Found
    作者

    你可以把循环的次数调大,它出错的机率会越高。