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

Question

作为我上一篇文章的后续 Is iteration via Collections.synchronizedSet(...).forEach() guaranteed to be thread safe?我将分享我认为是实现中的错误的想法，以验证它确实是错误。

我们有一个 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 excessive synchronization

...
To avoid liveness and safety failures, never cede control to the client within a synchronized method or block. ... From the perspective of the class with the synchronized region, such methods are alien. ... Depending on what an alien method does, calling it from a synchronized region can cause exceptions, deadlocks or data corruption.
...

(强调作者)

它看起来像 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:您正在打印集合，而无需等待添加线程完成其工作。因此你有一个竞争条件:添加线程可能已经完成，或者可能还没有开始。该错误在您的代码中，而不是在集合的代码中。