通过Collections.synchronizedSet(...).forEach()的迭代是否保证是线程安全的？

Question

我们知道,默认情况下迭代并发集合不是线程安全的,所以不能使用:

Set<E> set = Collections.synchronizedSet(new HashSet<>());
//fill with data
for (E e : set) {
    process(e);
}

这是因为在迭代期间可能会添加数据,因为没有排他锁set.

这在javadoc中描述Collections.synchronizedSet:

public static Set synchronizedSet(Set s)

返回由指定集支持的同步(线程安全)集.为了保证串行访问,必须通过返回的集完成对后备集的所有访问.

当迭代它时,用户必须手动同步返回的集合:

Set s = Collections.synchronizedSet(new HashSet());
...
synchronized (s) { Iterator i = s.iterator(); // Must be in the synchronized block while (i.hasNext()) foo(i.next()); }

不遵循此建议可能会导致非确定性行为.

然而,这并不适用于Set.forEach,它继承了默认的方法forEach从Iterable.forEach.

现在我查看了源代码,在这里我们可以看到我们有以下结构:

1. 我们要求一个Collections.synchronizedSet().

2. 我们得到一个:

   public static <T> Set<T> synchronizedSet(Set<T> s) {
       return new SynchronizedSet<>(s);
   }
   
   ...
   
   static class SynchronizedSet<E>
         extends SynchronizedCollection<E>
         implements Set<E> {
       private static final long serialVersionUID = 487447009682186044L;
   
       SynchronizedSet(Set<E> s) {
           super(s);
       }
       SynchronizedSet(Set<E> s, Object mutex) {
           super(s, mutex);
       }
   
       public boolean equals(Object o) {
           if (this == o)
               return true;
           synchronized (mutex) {return c.equals(o);}
       }
       public int hashCode() {
           synchronized (mutex) {return c.hashCode();}
       }
   }
   

3. 它扩展了SynchronizedCollection,其中明显的方法旁边有以下有趣的方法:

   // Override default methods in Collection
   @Override
   public void forEach(Consumer<? super E> consumer) {
       synchronized (mutex) {c.forEach(consumer);}
   }
   @Override
   public boolean removeIf(Predicate<? super E> filter) {
       synchronized (mutex) {return c.removeIf(filter);}
   }
   @Override
   public Spliterator<E> spliterator() {
       return c.spliterator(); // Must be manually synched by user!
   }
   @Override
   public Stream<E> stream() {
       return c.stream(); // Must be manually synched by user!
   }
   @Override
   public Stream<E> parallelStream() {
       return c.parallelStream(); // Must be manually synched by user!
   }
   

这里mutex使用的是与锁定的所有操作相同的对象Collections.synchronizedSet.

现在我们可以根据实现来判断它是否可以使用线程安全Collections.synchronizedSet(...).forEach(...),但它是否也符合规范的线程安全？

(令人困惑的是,Collections.synchronizedSet(...).stream().forEach(...)是不是由执行线程安全的,规范的判决似乎是未知以及.)

Answer 1

正如您所写的那样,通过实现来判断,JDK提供的集合forEach()是线程安全的(请参阅下面的免责声明),因为它需要监视要获取的互斥锁才能继续.

是否通过规范线程安全？

我的意见 - 不,这是一个解释.Collections.synchronizedXXX()用短语重写的javadoc说 - "所有方法都是线程安全的,除了那些用于迭代它的方法".

我的另一个,虽然非常主观的论点是yshavit写的 - 除非告知/读取,考虑API /类/任何不是线程安全的.

现在,让我们仔细看看javadocs.我想我可以声明该方法forEach()用于迭代它,因此,遵循javadoc的建议,我们应该认为它不是线程安全的,尽管它与现实(实现)相反.

无论如何,我同意yshavit的声明,即文档应该更新,因为这很可能是文档,而不是实现缺陷.但是,除了JDK开发人员之外,没有人可以肯定地说,请看下面的问题.

我想在这个讨论中提到的最后一点 - 我们可以假设自定义集合可以包装Collections.synchronizedXXX(),并且forEach()这个集合的实现可以是...可以是任何东西.该集合可能会对forEach()方法中的元素执行异步处理,为每个元素生成一个线程......它仅受作者想象力的限制,并且synchronized(互斥)换行不能保证此类情况的线程安全.该特定问题可能是不将forEach()方法声明为线程安全的原因.

Answer 2

这是值得看看的的文件Collections.synchronizedCollection,而不是Collections.synchronizedSet()作为该文件已经被清理:

当用户遍历它时,用户必须手动同步返回的集合  Iterator,Spliterator或者Stream:......

我认为,这很清楚,通过除了synchronized Collection本身之外的对象和使用其forEach方法之间的迭代之间存在区别.但即使使用旧的措辞,您也可以得出结论:存在这样的区别:

当迭代它时,用户必须手动同步返回的集合:...

^(我强调)

与以下文档相比Iterable.forEach:

对每个元素执行给定操作,Iterable直到处理完所有元素或操作引发异常.

虽然开发人员很清楚必须进行(内部)迭代才能实现这一点,但此迭代是一个实现细节.从给定规范的措辞来看,它只是一个(元)动作,用于对每个元素执行操作.

当使用该方法时,用户被不遍历元件,因此不负责中提到的同步Collections.synchronized…文档.

但是,这有点微妙,文档synchronizedCollection列出明确的手动同步的情况,我认为其他方法的文档也应该适应.