非阻塞队列，带批处理导向排水全部

Question

我们有一个专家、多生产者（用户）和单一消费者（引擎）队列。用户线程运行得更频繁，并且总是向队列中添加单个元素。引擎线程操作运行频率较低，并在批处理中处理堆栈元素。如果堆栈是空的，它将停止，直到用户线程添加了一个条目。这样，只有当队列从空变为1时才需要发出通知。

在此实现中，引擎线程不是一次迭代和删除一个项目，而是删除所有项目 - 一个 drainAll，而不是 drainTo。没有其他操作可以改变堆栈 - 只有用户线程添加，引擎线程 drainAll。

目前我们通过一个同步链表来做这件事，我们想知道是否有一个非阻塞的方式来做这件事。对JDK类的drainTo操作将迭代堆栈，我们只想在一次操作中获取堆栈中的所有内容，而不是迭代——因为每次迭代都会遇到与volatile/cas相关的逻辑，所以我们最好每次drainAll只遇到一次。引擎线程可以对每个单独的元素进行迭代和操作，而不涉及sync/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();
    }
}

asdf

Answer 1

目前我们通过同步链表来做到这一点，我们想知道是否有一种非阻塞的方法来做到这一点。JDK类上的drainTo操作将迭代堆栈，我们只想在一次操作中获取堆栈中的所有内容，而无需迭代

也许我不明白，但似乎使用 BlockingQueue.drainTo（...） 方法会比你的实现更好。例如，LinkedBlockingQueue.drainTo（...） 方法只有一个围绕该方法的锁 - 我没有看到迭代开销。

如果这不是一个学术讨论，那么我怀疑你的性能问题是队列本身的问题，应该把精力集中在其他方面。如果是学术性的，那么@Matt的答案可能会更好，尽管肯定要编写更多的代码来支持完整的< code>Collection方法列表。

Answer 2

堆栈具有非常简单的非阻塞实现，可轻松支持并发“全部弹出”操作，并且可以轻松检测空

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

在空旷的地方发信号-