记一次java8 parallelStream使用不当引发的血案

原创
2017/07/19 16:20
阅读数 5.3W

总所周知,Stream是Java 8 的一大亮点,很受开发人员的青睐, 其中包括笔者在内。Stream 大大增强了集合对象功能,它专注于对集合对象进行各种非常便利、高效的聚合操作,或者大批量数据操作。Stream API 借助于java8中新出现Lambda 表达式,极大的提高编程效率和程序可读性。so,还有什么理由拒绝使用呢?然而,这种不明真相的滥用,最终也会自食恶果。

有一天,收到邮件,线上环境抛出ArrayIndexOutOfBoundsException,部分异常堆栈如下

java.lang.ArrayIndexOutOfBoundsException
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) ~[?:1.8.0_77]
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62) ~[?:1.8.0_77]
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45) ~[?:1.8.0_77]
at java.lang.reflect.Constructor.newInstance(Constructor.java:423) ~[?:1.8.0_77]
at java.util.concurrent.ForkJoinTask.getThrowableException(ForkJoinTask.java:598) ~[?:1.8.0_77]
at java.util.concurrent.ForkJoinTask.reportException(ForkJoinTask.java:677) ~[?:1.8.0_77]
at java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:735) ~[?:1.8.0_77]
at java.util.stream.ForEachOps$ForEachOp.evaluateParallel(ForEachOps.java:160) ~[?:1.8.0_77]
at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateParallel(ForEachOps.java:174) ~[?:1.8.0_77]
at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233) ~[?:1.8.0_77]
at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418) ~[?:1.8.0_77]
at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:583) ~[?:1.8.0_77]

抛出异常的位置正是parallelStream()所在行,先贴出线上代码:

List<RiderDto> riderSubList = riderSearchProviderClient.getBatchRiderInfo(riderIdSub);
List<RiderBaseDto> subRiderBaseDTOs = Lists.newArrayList();
riderSubList.parallelStream().forEach(rider -> {//异常堆栈指示的位置
	RiderBaseDto subRiderBaseDTO = new RiderBaseDto();
	BeanUtils.copyProperties(rider, subRiderBaseDTO);
	subRiderBaseDTOs.add(subRiderBaseDTO);
});

虽然怀疑是parallelStream的问题,但是对其内部原理不甚了解,决定写个demo测试下,代码如下:

public class ParallelStreamTest {
	private static final int COUNT = 1000;
	public static void main(String[] args) {
		List<RiderDto> orilist=new ArrayList<RiderDto>();
        for(int i=0;i<COUNT;i++){
        	orilist.add(init());
        }
        final List<RiderDto> copeList=new ArrayList<RiderDto>();
        orilist.parallelStream().forEach(rider -> {
        	RiderDto t = new RiderDto();
        	t.setId(rider.getId());
    		t.setCityId(rider.getCityId());
        	copeList.add(t);
		});
        System.out.println("orilist size:"+orilist.size());
        System.out.println("copeList size:"+copeList.size());
        System.out.println("compare copeList and list,result:"+(copeList.size()==orilist.size())); 
	}
	private static RiderDto init() {
		RiderDto t = new RiderDto();
		Random random = new Random();
		t.setId(random.nextInt(2 ^ 20));
		t.setCityId(random.nextInt(1000));
		return t;
	}
	static class RiderDto implements Serializable{
		private static final long serialVersionUID = 1;
		//城市Id
	    private Integer cityId;
	    //骑手Id
	    private Integer id;
		......
	}
}

多次运行输出如下:

orilist size:1000
copeList size:998
compare copeList and orilist,result:false

orilist size:1000
copeList size:981
compare copeList and orilist,result:false

orilist size:1000
copeList size:1000
compare copeList and orilist,result:true
Exception in thread "main" java.lang.ArrayIndexOutOfBoundsException
	at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
	at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
	at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
	at java.util.concurrent.ForkJoinTask.getThrowableException(ForkJoinTask.java:598)
	at java.util.concurrent.ForkJoinTask.reportException(ForkJoinTask.java:677)
	at java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:735)
	at java.util.stream.ForEachOps$ForEachOp.evaluateParallel(ForEachOps.java:160)
	at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateParallel(ForEachOps.java:174)
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233)
	at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:418)
	at java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:583)
	at com.dianwoba.test.ParallelStreamTest.main(ParallelStreamTest.java:17)
Caused by: java.lang.ArrayIndexOutOfBoundsException: 244
	at java.util.ArrayList.add(ArrayList.java:459)
	at com.dianwoba.test.ParallelStreamTest.lambda$0(ParallelStreamTest.java:21)
	at java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
	at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1374)
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
	at java.util.stream.ForEachOps$ForEachTask.compute(ForEachOps.java:291)
	at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:731)
	at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
	at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
	at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
	at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)

结果让人很意外,每次输出的结果不一样,同时,确实抛出了异常ArrayIndexOutOfBoundsException,异常堆栈也和线上环境一样,由此断定是parallelStream使用不当造成的问题。下面探究下parallelStream的运行原理。

parallelStream是一个并行执行的流,其使用 fork/join (ForkJoinPool)并行方式来拆分任务和加速处理过程。研究parallelStream之前,搞清楚ForkJoinPool是很有必要的。

ForkJoinPool的核心是采用分治法的思想,将一个大任务拆分为若干互不依赖的子任务,把这些子任务分别放到不同的队列里,并为每个队列创建一个单独的线程来执行队列里的任务。同时,为了最大限度地提高并行处理能力,采用了工作窃取算法来运行任务,也就是说当某个线程处理完自己工作队列中的任务后,尝试当其他线程的工作队列中窃取一个任务来执行,直到所有任务处理完毕。所以为了减少线程之间的竞争,通常会使用双端队列,被窃取任务线程永远从双端队列的头部拿任务执行,而窃取任务的线程永远从双端队列的尾部拿任务执行。

到这里,我们知道parallelStream使用多线程并行处理数据,关于多线程,有个老生常谈的问题,线程安全。正如上面的分析,demo中orilist会被拆分为多个小任务,每个任务只负责处理一小部分数据,然后多线程并发地处理这些任务。问题就在于copeList不是线程安全的容器,并发调用add就会发生线程安全的问题,这里改用CopyOnWriteArrayList就不会有问题了。

final List<RiderDto> copeList=new CopyOnWriteArrayList<RiderDto>();

实际这里也没必要使用parallelStream,因此直接去掉parallelStream发到线上了。

那么,针对上面的输出结果,你就没有任何疑问么,又为什么copeList的长度会小?又为什么多线程调用ArrayList.add会发生数组越界异常呢?还是从源码解答吧。

    public boolean add(E e) {
        ensureCapacityInternal(size + 1); 
        elementData[size++] = e;
        return true;
    }

将size+1后调用ensureCapacityInternal确定ArrayList内部数组的容量。

   private void ensureCapacityInternal(int minCapacity) {
        if (elementData == DEFAULTCAPACITY_EMPTY_ELEMENTDATA) {
            minCapacity = Math.max(DEFAULT_CAPACITY, minCapacity);
        }

        ensureExplicitCapacity(minCapacity);
    }

如果当前数组为空,则DEFAULT_CAPACITY作为数组新的容量,继续跟踪ensureExplicitCapacity:

    private void ensureExplicitCapacity(int minCapacity) {
        modCount++;
        if (minCapacity - elementData.length > 0)
            grow(minCapacity);
    }

如果新的容量值大于数组的实际值,需要调用grow进行扩容。

    private void grow(int minCapacity) {
        int oldCapacity = elementData.length;
        int newCapacity = oldCapacity + (oldCapacity >> 1);
        if (newCapacity - minCapacity < 0)
            newCapacity = minCapacity;
        if (newCapacity - MAX_ARRAY_SIZE > 0)
            newCapacity = hugeCapacity(minCapacity);
        elementData = Arrays.copyOf(elementData, newCapacity);
    }

由实现可知,grow会自动扩容为原始容量的1.5倍,然后将原始数组中的元素重新拷贝一份到新的数组中,至此完成扩容。

相信到这里,都不是导致copeList长度会小的源头,真正发生问题的是elementData[size++] = e,解析这行代码,分解为几个原子操作:

  • 首先将e添加到size的位置,即elementData[size] = e
  • 读取size
  • size加1

由于这里存在内存可见性问题,当线程A从内存读取size后,将size加1,然后写入内存,过程中可能有线程B也修改了size并写入内存,那么线程A写入内存的值就会丢失线程B的更新,这也解释了为什么parallelStream运行完成后,会出现copeList的长度比原始数组要小的情况。

数组越界异常则主要发生在数组扩容前的临界点。下面开始分析:

假设当前数组刚好只能添加一个元素,两个线程同时进入: ensureCapacityInternal(size + 1),同时读取的size值,加1后进入ensureCapacityInternal都不会导致扩容,退出ensureCapacityInternal后,两个线程同时elementData[size] = e,线程B的size++先完成,假设此刻线程A读取到了线程B的更新,线程A再执行size++,此时size的实际值就会大于数组的容量,这样就会发生数组越界异常。

欢迎指出本文有误的地方,转载请注明原文出处https://my.oschina.net/7001/blog/1475500

展开阅读全文
打赏
3
1 收藏
分享
加载中
写得好!
2019/09/08 17:15
回复
举报
AbeJeffrey博主

引用来自“映日古月”的评论

可以的啊,通过riderSubList使用parallStream->map->tolist然后用subRiderBaseDTOs接收
👍 foreach是有副作用的,collect没有
2018/12/25 09:29
回复
举报
可以的啊,通过riderSubList使用parallStream->map->tolist然后用subRiderBaseDTOs接收
2018/12/24 17:58
回复
举报
同样踩到这个坑,使用parallelStream遍历list修改对象属性,发现有时候数据正常,有时候丢数据,还有时候返回list中包含null
2018/12/24 13:50
回复
举报
大哥,你用parallStream这是一个并发操作,你必须要使用线程安全的集合才可以
2018/12/21 10:38
回复
举报
更多评论
打赏
5 评论
1 收藏
3
分享
返回顶部
顶部