代码之家 › 专栏 › 技术社区 › Daniil Iaitskov

集合更新后的并行流

java-stream thread-safety multithreading java

Daniil Iaitskov · 技术社区 · 7 年前

我使代码与OpenJDk8兼容,并注意到有时下面的代码片段会引发ConcurrentModificationException。

代码在一个线程中执行。

list = new ArrayList<>();

list.addAll(newElements);
list.stream().parallel().forEach(o -> {});

当流线程在它们自己之间分割列表时,ForkJoin会引发异常。过了一会儿,我认为默认情况下这是合理的行为(不是同步的,简单的集合)流线程不是新的,它们是从池中使用的(它们可以看到处于不一致状态的列表),但是在迭代之前复制集合看起来不太好。

我还没有发现与此相关的bug,所以我问了一个问题:Java内存模型是如何看待这种情况的?


java.util.ConcurrentModificationException: java.util.ConcurrentModificationException
    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:593)
    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 scheduler.JobScheduleUpdater.fetchJobSchedules(JobScheduleUpdater.java:118)
    at scheduler.JobScheduleUpdater.processOneUpdate(JobScheduleUpdater.java:102)
    at 
scheduler.JobScheduleUpdater.execute(JobScheduleUpdater.java:81)
    at scheduler.JobScheduleUpdater$$EnhancerByMockitoWithCGLIB$$ccd35a20.CGLIB$execute$11(<generated>)
    at scheduler.JobScheduleUpdater$$EnhancerByMockitoWithCGLIB$$ccd35a20$$FastClassByMockitoWithCGLIB$$bf9de679.invoke(<generated>)
    at org.mockito.cglib.proxy.MethodProxy.invokeSuper(MethodProxy.java:216)
    at org.powermock.api.mockito.repackaged.DelegatingMockitoMethodProxy.invokeSuper(DelegatingMockitoMethodProxy.java:20)
    at org.mockito.internal.invocation.realmethod.DefaultRealMethod.invoke(DefaultRealMethod.java:21)
    at org.mockito.internal.invocation.realmethod.CleanTraceRealMethod.invoke(CleanTraceRealMethod.java:30)
    at org.mockito.internal.invocation.InvocationImpl.callRealMethod(InvocationImpl.java:112)
    at org.mockito.internal.stubbing.answers.CallsRealMethods.answer(CallsRealMethods.java:41)
    at org.mockito.internal.handler.MockHandlerImpl.handle(MockHandlerImpl.java:93)
    at org.mockito.internal.handler.NullResultGuardian.handle(NullResultGuardian.java:29)
    at org.mockito.internal.handler.InvocationNotifierHandler.handle(InvocationNotifierHandler.java:38)
    at org.powermock.api.mockito.repackaged.MethodInterceptorFilter.intercept(MethodInterceptorFilter.java:60)
    at scheduler.JobScheduleUpdater$$EnhancerByMockitoWithCGLIB$$ccd35a20.execute(<generated>)
    at commons.util.service.ShutdownCapableThread.run(ShutdownCapableThread.java:74)
    at scheduler.JobScheduleUpdater$$EnhancerByMockitoWithCGLIB$$ccd35a20.CGLIB$run$15(<generated>)
    at scheduler.JobScheduleUpdater$$EnhancerByMockitoWithCGLIB$$ccd35a20$$FastClassByMockitoWithCGLIB$$bf9de679.invoke(<generated>)
    at org.mockito.cglib.proxy.MethodProxy.invokeSuper(MethodProxy.java:216)
    at org.powermock.api.mockito.repackaged.DelegatingMockitoMethodProxy.invokeSuper(DelegatingMockitoMethodProxy.java:20)
    at org.mockito.internal.invocation.realmethod.DefaultRealMethod.invoke(DefaultRealMethod.java:21)
    at org.mockito.internal.invocation.realmethod.CleanTraceRealMethod.invoke(CleanTraceRealMethod.java:30)
    at org.mockito.internal.invocation.InvocationImpl.callRealMethod(InvocationImpl.java:112)
    at org.mockito.internal.stubbing.answers.CallsRealMethods.answer(CallsRealMethods.java:41)
    at org.mockito.internal.handler.MockHandlerImpl.handle(MockHandlerImpl.java:93)
    at org.mockito.internal.handler.NullResultGuardian.handle(NullResultGuardian.java:29)
    at org.mockito.internal.handler.InvocationNotifierHandler.handle(InvocationNotifierHandler.java:38)
    at org.powermock.api.mockito.repackaged.MethodInterceptorFilter.intercept(MethodInterceptorFilter.java:60)
    at scheduler.JobScheduleUpdater$$EnhancerByMockitoWithCGLIB$$ccd35a20.run(<generated>)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.util.ConcurrentModificationException
    at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1388)
    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)

1 回复 | 直到 7 年前

BMasonJ13 7 年前

代码在我的编辑器中运行良好。这个 ConcurrentModificationException 当您修改正在迭代的内容时触发。如果没有看到你所有的代码,我想我无法解决这个问题。

一个例子 ConcurrentModificationException 是:

List<Integer> integers = new ArrayList<>();

integers.add(12);
integers.add(11);
integers.add(1);



for (Integer integer : integers) {
   integers.remove(1);
}

可以找到有关如何解决问题的资源 here