java - 如何设计一个互斥但独立并发方法的任务队列？

Question

我在一次工作面试中被问到了一个并发问题，最终归结为以下要求。我仅通过使用互斥就能实现#2 到#4，但不能实现#1。

Design a task queue with the following methods:

public void registerCallback(Runnable task)

public void eventFired()

1. Multiple threads should be able to put tasks on the queue, possibly concurrently.
2. eventFired should only be invoked once.
3. If eventFired has been previously invoked, any later invocation of either methods should throw an exception.
4. If eventFired is invoked while registerCallback is executing, delay firing the event until a later time.
5. If registerCallback is invoked while eventFired is executing, throw an exception.

ReentrantReadWriteLock 看起来很有前途，因为 registerCallback 可以获取读锁，而 eventFired 可以获取写锁，但这并不能解决竞争问题调用 registerCallback 的条件，然后调用 eventFired。

有什么想法吗？

Answer 1

ReentrantReadWriteLock seems promising, because registerCallback can acquire the read lock, and eventFired the write lock, but that doesn't solve a race condition where registerCallback is invoked, and then eventFired.

注册回调，换句话说，在持有读锁的同时修改数据结构从来都不是一个好主意。您需要另一个线程安全的构造来存储回调，并使代码不必要地复杂化。

注册回调的操作(例如将引用存储到某个集合中或实例化任何类型的节点对象)非常简单，足以允许使用普通的互斥锁，因为它不会保持很长时间。

无论您使用synchronized、Lock还是支持原子更新的状态，在任何一种情况下，竞争条件都不存在，因为不可能存在registerCallback 或 eventFired 的重叠。如果使用得当，所有这些方法都会给这些操作带来秩序。因此，每个 registerCallback 要么在第一个 eventFired 之前，要么在它之后。

实现这一点可以很简单:

public class JobQueue {
    private List<Runnable> callbacks = new ArrayList<>();

    public synchronized void registerCallback(Runnable task) {
        if(callbacks == null) {
            throw new IllegalStateException("Event already fired");
        }
        callbacks.add(Objects.requireNonNull(task));
    }
    public void eventFired() {
        List<Runnable> list;
        synchronized(this) {
            list = callbacks;
            callbacks = null;
        }
        if(list == null) {
            throw new IllegalStateException("Can only fire once");
        }
        for(Runnable r: list) r.run();
    }
}

在synchronized block 中执行的代码非常短，因此对于大多数实际用例来说，争用是无关紧要的。使用 Lock 实现相同的功能会很简单，但不会有任何优势。事实上，JVM 特定的优化可能会使基于同步的解决方案更加高效。

为了完整起见，这里是一个基于原子更新的解决方案:

public class JobQueue {
    private static final Runnable DONE = () -> {};

    private final AtomicReference<Runnable> pending = new AtomicReference<>();

    public void registerCallback(Runnable task) {
        Objects.requireNonNull(task);
        for(;;) {
            Runnable previous = pending.get();
            if(previous == DONE) throw new IllegalStateException("Event already fired");
            if(pending.compareAndSet(previous, sequence(previous, task))) return;
        }
    }

    public void eventFired() {
        Runnable previous = pending.getAndSet(DONE);
        if(previous == DONE) throw new IllegalStateException("Can only fire once");
        if(previous != null) previous.run();
    }

    static Runnable sequence(Runnable a, Runnable b) {
        return a == null? b: () -> { a.run(); b.run(); };
    }
}

实际上，多个 registerCallback 和/或 eventFired 调用的执行可能会重叠，但在这种情况下，只有一个可以成功执行关键的原子更新。这会给操作带来顺序，最多使一次 eventFired 调用成功，并将所有 registerCallback 调用分类为之前或之后那个。