java - 检查大小然后执行操作 - 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

就内部 状态而言，并发集合是线程安全的。换句话说，他们

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

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

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

可能会发生以下情况:

1. Thread-A 检查大小条件，结果为 false
2. Thread-A 移动添加新的Operation
3. 在 Thread-A 可以添加 Operation 之前，Thread-B 检查导致 false 的大小条件还有
4. Thread-B 去添加新的Operation
5. Thread-A does 添加新的Operation
   • 哦，不! Thread-A 添加的Operation 导致达到大小阈值
6. Thread-B，已经通过了 if 语句，添加了它的 Operation 使得双端队列有一个太多的 Operations

这就是为什么 check-then-act 需要外部同步，您在第二个示例中使用 Lock 来执行此操作。请注意，您还可以在 Deque 上使用 synchronized block 。

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

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