检查大小然后执行操作 - ConcurrentLinkedDeque 是否安全？

Question

我需要用新值替换 Deque 中的第一个值，仅如果大小将超过限制。我写了这段代码来解决它：

final class Some {
    final int buffer;
    final Deque<Operation> operations = new ConcurrentLinkedDeque<>();
    // constructors ommited;

    @Override
    public void register(final Operation operation) {
        if (this.operations.size() == this.buffer) {
            // remove the oldest operation
            this.operations.removeFirst();
        }
        // add new operation to the tail
        this.operations.addLast(operation);
    }

    @Override
    public void apply() {
        // take the fresh operation from tail and perform it
        this.operations.removeLast().perform();
    }
}

如您所见，我有两种方法可以修改 Deque。我怀疑这段代码能否在多线程环境中正常工作。问题是：检查 size() 然后执行操作，然后修改 ConcurrentLinkedDeque 是否安全？我想拥有尽可能少的锁。因此，如果此代码不起作用，那么我必须引入锁定，然后 ConcurrentLinkedDeque().

的用法就没有意义了

final class Some {
    final int buffer;
    final Deque<Operation> operations = new LinkedList<>();
    final Lock lock = new ReentrantLock();
    // constructors ommited;

    @Override
    public void register(final Operation operation) {
        this.lock.lock();
        try {
            if (this.operations.size() == this.buffer) {
                // remove the oldest operation
                this.operations.removeFirst();
            }
            // add new operation to the tail
            this.operations.addLast(operation);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void apply() {
        this.lock.lock();
        try {
            // take the fresh operation from tail and perform it
            this.operations.removeLast().perform();
        } finally {
            this.lock.unlock();
        }
    }
}

这是 Lock 的备选方案。这是实现我想要的唯一方法吗？我对尝试使用并发集合特别感兴趣。

Answer 1

并发集合在内部状态下是线程安全的。也就是说，他们

允许多个线程并发read/write而不必担心内部状态会被破坏
允许在其他线程修改集合时进行迭代和删除
- 然而，并非全部。我相信CopyOnWriteArrayList的Iterator不支持remove()操作
保证诸如 happens-before 之类的事情
- 意味着一个线程的写入将发生在后续线程的读取

但是，它们不是线程安全的跨外部方法调用。当您调用一个方法时，它将获得任何必要的锁，但这些锁会在方法 returns 时释放。如果您不小心，这可能会导致 check-then-act 竞争条件。查看您的代码

if (this.operations.size() == this.buffer) {
    this.operations.removeFirst();
}
this.operations.addLast(operation);

可能会发生以下情况：

Thread-A 检查尺寸条件，结果为 false
Thread-A移动添加新的Operation
在 Thread-A 可以添加 Operation 之前，Thread-B 检查大小条件，这也会导致 false
Thread-B 去添加新的 Operation
Thread-A 是否添加新的 Operation
- 哦，不！ Thread-A 添加的 Operation 导致达到大小阈值
Thread-B，已经过了 if 语句，添加它的 Operation 使得双端队列有一个太多 Operations

这就是 check-then-act 需要外部同步的原因，您在第二个示例中使用了 Lock。请注意，您还可以在 Deque.

上使用 synchronized 块

与您的问题无关：您在第二个示例中调用 Operation.perform()，同时仍按住 Lock。这意味着当 perform() 执行时，没有其他线程可以尝试将另一个 Operation 添加到 Deque。如果不需要，您可以像这样更改代码：

Operation op;

lock.lock();
try {
    op = deque.pollLast(); // poll won't throw exception if there is no element
} finally {
    lock.unlock();
}

if (op != null) {
    op.perform();
}

Answer 2

来自 size() 的文档

BlockquoteBeware that, unlike in most collections, this method is NOT a constant-time operation. Because of the asynchronous nature of these deques, determining the current number of elements requires traversing them all to count them. Additionally, it is possible for the size to change during execution of this method, in which case the returned result will be inaccurate. Thus, this method is typically not very useful in concurrent applications.

虽然@Slaw 是正确的，但还要补充一点，addition/subtraction 可能会在遍历过程中发生。

我的软件中没有使用 size()。我用 AtomicInteger 自己计算集合中的内容。如果 count.get() < max，我可以添加。稍微超过 max 对我的使用来说是可以的。您可以使用计数锁定来强制合规。

检查大小然后执行操作 - ConcurrentLinkedDeque 是否安全？

Check size and then perform operation - is it safe for ConcurrentLinkedDeque?

java

concurrency

multithreading

deque