Collections.synchronizedCollection的forEach和removeIf中可能存在错误？

Question

作为我上一篇文章的后续内容是通过Collections.synchronizedSet(...)进行迭代.forEach()保证是线程安全的吗？我将分享我对我认为实施中的错误的看法,以验证它确实是一个错误.

我们有一个SynchronizedCollection<E>这里,可以通过调用Collections.synchronizedCollection(...)JDK来源获得:

public static <T> Collection<T> synchronizedCollection(Collection<T> c) {
    return new SynchronizedCollection<>(c);
}

---

static class SynchronizedCollection<E> implements Collection<E>, Serializable {
    private static final long serialVersionUID = 3053995032091335093L;

    final Collection<E> c;  // Backing Collection
    final Object mutex;     // Object on which to synchronize

    SynchronizedCollection(Collection<E> c) {
        this.c = Objects.requireNonNull(c);
        mutex = this;
    }

    SynchronizedCollection(Collection<E> c, Object mutex) {
        this.c = Objects.requireNonNull(c);
        this.mutex = Objects.requireNonNull(mutex);
    }

    public int size() {
        synchronized (mutex) {return c.size();}
    }
    public boolean isEmpty() {
        synchronized (mutex) {return c.isEmpty();}
    }
    public boolean contains(Object o) {
        synchronized (mutex) {return c.contains(o);}
    }
    public Object[] toArray() {
        synchronized (mutex) {return c.toArray();}
    }
    public <T> T[] toArray(T[] a) {
        synchronized (mutex) {return c.toArray(a);}
    }

    public Iterator<E> iterator() {
        return c.iterator(); // Must be manually synched by user!
    }

    public boolean add(E e) {
        synchronized (mutex) {return c.add(e);}
    }
    public boolean remove(Object o) {
        synchronized (mutex) {return c.remove(o);}
    }

    public boolean containsAll(Collection<?> coll) {
        synchronized (mutex) {return c.containsAll(coll);}
    }
    public boolean addAll(Collection<? extends E> coll) {
        synchronized (mutex) {return c.addAll(coll);}
    }
    public boolean removeAll(Collection<?> coll) {
        synchronized (mutex) {return c.removeAll(coll);}
    }
    public boolean retainAll(Collection<?> coll) {
        synchronized (mutex) {return c.retainAll(coll);}
    }
    public void clear() {
        synchronized (mutex) {c.clear();}
    }
    public String toString() {
        synchronized (mutex) {return c.toString();}
    }
    // 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!
    }
    private void writeObject(ObjectOutputStream s) throws IOException {
        synchronized (mutex) {s.defaultWriteObject();}
    }
}

现在让我们再看一下这段代码:

    // 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);}
    }

它们是此实现中允许插入任何代码的唯一方法.

现在我将控制权转移到Effective Java:Item 67:Avid过度同步

...
为避免活动和安全故障,请勿在同步方法或块中将控制权交给客户端.......从具有同步区域的类的角度来看,这种方法是陌生的....根据异类方法的作用,从同步区域调用它可能会导致异常,死锁或数据损坏.
...

(重点作者)

看起来这些forEach和removeIf方法Collections.synchronizedCollection完全违反了该规则,我认为它在Java 8中仍然有效.

所以我们应该能够在这里构造两个死锁的SSCEE.

我将此标记为可能的错误的原因之一是,我无法使用以下代码重现它:

public class Java8BugSSCEE1 {
    public static void main(String[] args) {
        Collection<String> collection = Collections.synchronizedCollection(new HashSet<>());
        collection.add("Test");
        collection.forEach(str -> {
            synchronized (collection) {
                System.out.println("Obtained lock");
                collection.add(str + Integer.toHexString(ThreadLocalRandom.current().nextInt(16)));
            }
        });
        System.out.println("collection = " + collection);
    }
}

这[Test, TestX]按预期打印,X in [0, f]以十六进制表示.

我的想法是:

1. SynchronizedCollection.forEach将锁定mutex,这正是返回collection.
2. 然后在我的lambda内我会尝试获得相同的锁并失败.

但是最后一步没有发生,并且该程序正常执行.

所以最后一个问题,Java 8实现中是否存在错误？

编辑,我有新的SSCEE,可能确实显示了实现中的错误.

public class Java8BugSSCEE1 {
    public static void main(String[] args) {
        Collection<String> collection = Collections.synchronizedCollection(new HashSet<>());
        collection.add("Test");
        collection.forEach(str -> {
            new Thread(() -> {
                collection.add(str + Integer.toHexString(ThreadLocalRandom.current().nextInt(16)));
            }).start();
        });
        System.out.println("collection = " + collection);
    }
}

我希望这个版本添加一个元素,但它不添加任何东西.

Answer 1

这不是一个错误.这是一个功能.synchronizedCollection()返回的集合使其所有方法都同步,这就是全部.这是记录在案的行为.

这会自动使您的代码线程安全吗？没有.

这是否可以防止任何死锁的可能性？没有.

调用者是否应该了解他正在采取哪些措施来防止死锁并使其代码具有线程安全性？当然.

为什么你的代码不会失败？因为锁是可重入的.您可以让一个线程多次锁定同一个对象而不会出现问题.这就是你的代码所做的.要有死锁,您需要多个线程.

编辑:

关于你的最后一次SSCCE:你正在打印集合而不等待添加线程完成它的工作.因此,您有一个竞争条件:添加线程可能已完成,或者尚未启动.错误在您的代码中,而不在集合的代码中.