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

2022-09-01 13:25:04

众所周知,默认情况下,迭代并发集合不是线程安全的,因此不能使用:

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

发生这种情况是因为在迭代期间可能会添加数据,因为 上没有独占锁。set

这在以下的 javadoc 中进行了描述:Collections.synchronizedSet

公共静态 设置同步集(集)

返回由指定集支持的同步(线程安全)集。为了保证串行访问,对支持集的所有访问都必须通过返回的集合完成,这一点至关重要。

用户在迭代返回的集合时必须手动同步该集:

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

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

但是,这不适用于 ,它继承了 Iterable.forEach 的默认方法。Set.forEachforEach

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

  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!
    }
    

这里使用的对象与所有操作锁定到的对象相同。mutexCollections.synchronizedSet

现在我们可以,从实现判断说它是线程安全的 使用 ,但是按规范也可以线程安全吗Collections.synchronizedSet(...).forEach(...)

(令人困惑的是,通过实现不是线程安全的,并且规范的结论似乎也是未知的。Collections.synchronizedSet(...).stream().forEach(...)


答案 1

正如你所写的,从实现来看,对于随 JDK 提供的集合(请参阅下面的免责声明)是线程安全的,因为它需要获取互斥体的监视器才能继续。forEach()

根据规格,它是否也是线程安全的?

我的观点-不,这里有一个解释。 javadoc,用简短的文字重写,说 - “除了那些用于迭代它的方法之外,所有方法都是线程安全的”。Collections.synchronizedXXX()

我的另一个,虽然非常主观的论点是yshavit写的 - 除非被告知/阅读,否则请考虑API / class /任何不是线程安全的。

现在,让我们仔细看看javadocs。我想我可能会说方法用于迭代它,所以,按照javadoc的建议,我们应该认为它不是线程安全的,尽管它与现实(实现)相反。forEach()

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

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


答案 2

值得一看 Collections.synchronizedCollection 的文档,而不是因为文档已经清理过了:Collections.synchronizedSet()

用户在通过 或 : ...IteratorSpliteratorStream

我认为,这清楚地表明,通过同步本身以外的对象进行迭代和使用其方法之间存在差异。但即使使用旧的措辞,您也可以得出这样的结论:存在这样的区别:CollectionforEach

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

(由我强调)

Iterable.forEach 的文档进行比较:

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

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

使用该方法时,用户不会循环访问元素,因此不负责文档中提到的同步。Collections.synchronized…

但是,这有点微妙,并且同步收集的文档明确列出了手动同步的情况是件好事,我认为其他方法的文档也应该进行调整。