在 anyMatch 调用期间流式传输 HashSet 时出现 ConcurrentModificationException

Question

    private Set<Job> myJobs = new HashSet<>();

    public shouldDoWork(Work work)
        return !myJobs.stream()
                  .map(job -> job.doWork(work))
                  .anyMatch(shouldDoWork -> !shouldDoWork);


   public addJob(Job job) {
       myJobs.add(job);
   }
   // also for remove

而且随时都有很多线程调用这个函数

java.util.ConcurrentModificationException 在 java.util.HashMap$KeySpliterator.tryAdvance(HashMap.java:1579) ~[?:1.8.0_202] 在 java.util.stream.ReferencePipeline.forEachWithCancel(ReferencePipeline.java:126) ~[?:1.8.0_202] 在 java.util.stream.AbstractPipeline.copyIntoWithCancel(AbstractPipeline.java:498) ~[?:1.8.0_202] 在 java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:485) ~[?:1.8.0_202] 在 java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471) ~[?:1.8.0_202] 在 java.util.stream.MatchOps$MatchOp.evaluateSequential(MatchOps.java:230) ~[?:1.8.0_202] 在 java.util.stream.MatchOps$MatchOp.evaluateSequential(MatchOps.java:196) ~[?:1.8.0_202] 在 java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:1.8.0_202] 在 java.util.stream.ReferencePipeline.anyMatch(ReferencePipeline.java:449) ~[?:1.8.0_202] 在

知道为什么会抛出 ConcurrentModificationException 吗？

是因为集合 (myJobs) 正在改变，还是因为 shouldDoWork 的值被其他东西改变了？

Answer 1

知道为什么会抛出 ConcurrentModificationException 吗？

是的。

是因为集合 (myJobs) 正在改变，还是因为 shouldDoWork 正在被其他东西改变吗？

前者。您列出了两种方法，shouldDoWork() 和 addJob()。第一个对集合myJobs 执行流操作，第二个向该集合添加一个元素。如果确实

任何时候都有很多线程调用这个函数

那么很可能在某个时候一个线程将调用addJob()，从而在结构上修改 set myJobs，在另一个从那个集合构造一个流的时间和另一个完成的时间之间消耗该流。修改集合中的对象不会导致ConcurrentModificationException（尽管如果这样做会更改元素的哈希码，它通常会使集合无效）。它正在修改集合本身可以做的。

事实上，您很幸运，您获得了 CME，因为在您描述的不正确同步修改的情况下，这并不能保证。你可能会得到一个垃圾结果。

对于两者实现正确同步并避免 CME，您至少有两个可行的替代方案：

1. 同步访问myJobs。例如，

   public boolean shouldDoWork(Work work) {
       synchronized(myJobs) {
           return !myJobs.stream()
                 .map(job -> job.doWork(work))
                 .anyMatch(shouldDoWork -> !shouldDoWork);
       }
   }
   
   public addJob(Job job) {
       synchronized (myJobs) {
           myJobs.add(job);
       }
   }
   

   如果您采用这种方法，那么您必须同样将所有访问同步到myJobs。或

2. 使用不同类型的容器，例如 ConcurrentHashMap，不需要显式同步

   private ConcurrentHashMap<Job, Job> myJobs = new ConcurrentHashMap<>();
   
   public boolean shouldDoWork(Work work) {
       return !myJobs.keySet().stream()
             .map(job -> job.doWork(work))
             .anyMatch(shouldDoWork -> !shouldDoWork);
   }
   
   public addJob(Job job) {
       myJobs.put(job, job);
   }
   

   在实施此类更改之前，您应该阅读建议的替代类的文档（在这种情况下为ConcurrentHashMap），以确保您了解其中的含义。这种替代方案可能比同步执行得更好，但这是以较弱的行为保证为代价的。