java - 使用两个单独的 Java 并发结构重新排序

Question

我正在对项目队列启动一个长时间运行的进程，当某个项目已计划处理或正在处理时，我想禁止某些其他操作。我的代码基本上是这样的:

public class LongRunningProcess extends Thread {
    private final ConcurrentLinkedQueue<Item> pending = new ConcurrentLinkedQueue<>();
    private final Set<Item> active = Collections.newSetFromMap(new ConcurrentHashMap<Item, Boolean>());

    public LongRunningProcess() {
        // add items to pending; no more items will ever be added
    }

    @Override
    public void run() {
        while (! pending.isEmpty()) {
            // The peek/add/remove pattern here is important. The number
            // of items that are active or scheduled is always decreasing.
            // Because isScheduled checks pending before checking active,
            // this order of operations ensures that we never miss an item
            // as it is being switched from one collection to the other.
            Item nextItem = pending.peek();
            active.add(nextItem);    // <---Can any of these get reordered?
            pending.remove();        // <---+
            processItem(nextItem);   // <---+
            active.remove(nextItem); // <---+
        }
    }

    public boolean isScheduled(Item item) {
        return pending.contains(item) || active.contains(item);
    }
}

这会按照我期望的方式工作吗？或者上面突出显示的代码块是否可以重新排序？您能给我指出任何相关规范吗？

编辑:

@Banthar 的有用评论让我找到了 java.util.concurrent package documentation ，这明确地回答了我的问题:

The methods of all classes in java.util.concurrent and its subpackages extend these guarantees to higher-level synchronization. In particular:

• Actions in a thread prior to placing an object into any concurrent collection happen-before actions subsequent to the access or removal of that element from the collection in another thread.

Answer 1

Will this work the way I expect, or is it possible for either of the two highlighted items above to be reordered? Can you please point me to any relevant specs?

简短的回答是，因为两个集合都是并发类，所以 active.add(...) 不可能在 pending.remove() 之后发生。

• pending.peek(); 和 pending.remove(); 访问 volatile 字段 head。

  private transient volatile Node<E> head = new Node<E>(null);
  

• active.add(nextItem); 访问内部锁定 volatile 字段:

  compareAndSetState(0, acquires)) {
  

因为您的两个集合都是并发类，所以它们都有内部锁或 volatile 变量，因此方法调用具有读/写内存屏障，以确保“发生在之前”保证。这确保了操作不会因为Java Memory Model而被重新排序。 .

但是，这并不意味着您的逻辑是正确的，或者当您查看其他线程如何使用这两个集合时不存在竞争条件。此外，这些调用不是原子的，因此您可以有 3 个线程执行以下操作:

1. t1 -- 项目 nextItem =ending.peek();
2. t2 -- 项目 nextItem =ending.peek();
3. t1 -- active.add(nextItem);
4. t3 -- 从 Activity 状态中删除 nextItem 并对其进行处理
5. t2 -- active.add(nextItem);
6. t3 -- 从 Activity 状态中删除 nextItem 并再次处理