java - 竞争激烈的消费者-生产者

Question

我想知道实现多个生产者-单个消费者场景的最佳机制是什么，在该场景中我必须保持当前未处理请求的数量最新。

我的第一个想法是使用 ConcurrentLinkedQueue:

public class SomeQueueAbstraction {

    private Queue<SomeObject> concurrentQueue = new ConcurrentLinkedQueue<>();
    private int size;

    public void add(Object request) {
        SomeObject object = convertIncomingRequest(request);   
        concurrentQueue.add(object);
        size++;

    }

    public SomeObject getHead() {
        SomeObject object = concurrentQueue.poll();
        size--;
    }

    // other methods

问题是我必须在 add 上显式同步和size ++ ，以及 poll和size-- ，始终准确size这使得 ConccurentLinkedQueue一开始就没有意义。

在保持数据一致性的同时实现尽可能好的性能的最佳方法是什么？

我应该使用 ArrayDequeue相反并显式同步或者有更好的方法来实现这一点？

这里有类似的问题/答案:

java.util.ConcurrentLinkedQueue

其中讨论了如何对 ConcurrentLinkedQueue 进行复合操作自然不是原子的，但没有直接答案什么是给定场景的最佳选择。

注意:我明确计算大小，因为固有的 .size() 方法的时间复杂度是 O(n)。

Note2:我还担心我没有明确编写的 getSize() 方法会增加更多的争用开销。调用得比较频繁。

我正在寻找处理多个生产者的最有效方法 - 频繁调用 getSize() 的单个消费者。

替代建议:如果 SomeObject 结构中有 elementId，我可以从 ConcurrentLinkedQueue.poll() 获取当前大小，并且只需在机制内完成锁定即可生成此类 id。现在可以正确使用 Add 和 get，无需额外锁定。作为替代方案，效果如何？

Answer 1

因此，要求是报告最新的未处理请求数。而且经常要求这样做，这确实使得 ConcurrentLinkedQueue.size() 不适合。

这可以使用 AtomicInteger 来完成:它速度很快，并且总是尽可能接近当前未处理请求的数量。

这是一个示例，请注意一些小更新以确保报告的大小准确:

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class SomeQueueAbstraction {

    private final Queue<SomeObject> concurrentQueue = new ConcurrentLinkedQueue<>();
    private final AtomicInteger size = new AtomicInteger();

    public boolean add(Object request) {

        SomeObject object = convertIncomingRequest(request);   
        if (concurrentQueue.add(object)) {
            size.incrementAndGet();
            return true;
        }
        return false;
    }

    public SomeObject remove() {

        SomeObject object = concurrentQueue.poll();
        if (object != null) {
            size.decrementAndGet();
        }
        return object;
    }

    public int getSize() { return size.get(); }

    private SomeObject convertIncomingRequest(Object request) { 
        return new SomeObject(getSize()); 
    }

    class SomeObject {
        int id;
        SomeObject(int id) { this.id = id; }
    }
}