java - 面向批处理排水的非阻塞队列

Question

我们有一个专家，多生产者(用户)和单消费者(引擎)队列。用户线程运行频率更高，并且总是将单个元素添加到队列中。 Engine线程操作的运行频率降低，并批量处理堆栈元素。如果堆栈为空，它将停放，直到User线程添加了一个条目。这样，仅当队列从空变为1时才需要发生通知。

在此实现中，代替一次一次迭代并删除一个引擎项的引擎线程，它删除了所有这些项-raineAll，而不是drainTo。没有其他操作可以使堆栈发生变化-仅添加User线程，并删除引擎线程raineAll。

当前，我们通过同步链表进行此操作，我们想知道是否存在非阻塞方法。在JDK类上的raintTo操作将迭代堆栈，我们只想一次执行堆栈中的所有内容，而无需进行迭代-每次迭代都命中volatile/cas相关逻辑，因此理想情况下，我们希望每次执行一次全部清空。引擎线程可以迭代并在每个单独的元素上操作，而无需进行同步/ volatile /cas操作。

当前的实现如下所示:

public class SynchronizedPropagationQueue implements PropagatioQueue {
    protected volatile PropagationEntry head;
    protected volatile PropagationEntry tail;

    protected synchronized void addEntry( PropagationEntry entry ) {
        if ( head == null ) {
            head = entry;
            notifyWaitOnRest();
        } else {
            tail.setNext( entry );
        }
        tail = entry;
    }

    @Override
    public synchronized PropagationEntry drainAll() {
        PropagationEntry currentHead = head;
        head = null;
        tail = null;
        return currentHead;
    }

    public synchronized void waitOnRest() {
        try {
            log.debug("Engine wait");
            wait();
        } catch (InterruptedException e) {
            // do nothing
        }
        log.debug("Engine resumed");
    }


    @Override
    public synchronized void notifyWaitOnRest() {
        notifyAll();
    }
}

自卫队

Answer 1

堆栈具有非常简单的非阻塞实现，可轻松支持并发的“全部弹出”操作，并且可以轻松检测到空->非空过渡。您可以让所有生产者将物品插入堆栈，然后让引擎立即将整个物品清空。看起来像这样:

public class EngineQueue<T>
{
    private final AtomicReference<Node<T>> m_lastItem = new AtomicReference<>();

    public void add(T item)
    {
        Node<T> newNode = new Node<T>(item);
        do {
            newNode.m_next = m_lastItem.get();
        } while(!m_lastItem.compareAndSet(newNode.m_next, newNode)); 

        if (newNode.m_next == null)
        {
            // ... just went non-empty signal any waiting consumer
        }
    }

    public List<T> removeAll()
    {
        Node<T> stack = m_lastItem.getAndSet(null);
        // ... wait for non-empty if necessary 
        List<T> ret = new ArrayList<>();
        for (;stack != null; stack=stack.m_next)
        {
            ret.add(stack.m_data);
        }
        Collections.reverse(ret);
        return ret;
    }
    private static class Node<U>
    {
        Node<U> m_next;
        final U m_data;
        Node(U data)
        {
            super();
            m_data = data;
        }
    }
}

要在空->非空过渡周围发信号，可以使用普通同步。如果仅在检测到空状态时才这样做，这将不会很昂贵...因为您只有在没有工作时才能进入空状态。