java - ArrayBlockingQueue，如果在添加元素时队列已满，则删除队列头

Question

我正在尝试编写一个像ArrayBlockingQueue这样的简单队列，其中如果在添加元素时队列已满，则队列的头部将被删除。该类应该只具有以下公共(public)方法

• 获取队列的大小
• 从队列头部获取一个元素。如果没有可用元素则阻止。
• 在队列尾部添加元素

有人可以查看下面的代码并让我知道是否有更好的方法吗？

public class CircularArrayNonBlockingQueue<E> {
    private ArrayBlockingQueue<E> blockingQueue;

    public CircularArrayNonBlockingQueue(int size) {
        blockingQueue = new ArrayBlockingQueue<>(size);
    }

    public synchronized int size() {
        return blockingQueue.size();
    }

    public synchronized void add(E element) {
        if(blockingQueue.remainingCapacity() <= 0) {
            blockingQueue.poll();
        }
        blockingQueue.add(element);
    }

    public synchronized E poll() {
        return blockingQueue.poll();
    }
}

编辑根据评论中的讨论，我不需要使所有方法同步。更新后的代码如下所示 -

public class CircularNonBlockingQueue<E> {
    private final ArrayBlockingQueue<E> blockingQueue;

    public CircularNonBlockingQueue(int size) {
        blockingQueue = new ArrayBlockingQueue<>(size);
    }

    public int size() {
        return blockingQueue.size();
    }

    public synchronized void add(E element) {
        if(blockingQueue.remainingCapacity() <= 0) {
            blockingQueue.poll();
        }
        blockingQueue.add(element);
    }

    public E take() throws InterruptedException {
        return blockingQueue.take();
    }
}

Answer 1

拥有线程安全的后端集合并不一定能生成正确的程序。当只有你的add方法是synchronized ，take()方法可能会与其同时运行，因此有可能在您的 if(blockingQueue.remainingCapacity() <= 0) 之后在add内进行测试，同时运行的take()删除一个元素，因此 poll() add内可能会不必要地删除某个元素。 add() 的情况存在明显的差异。将在 take() 之前完成，因为消费线程将收到不同的项目。换句话说，效果就像 add有时不会删除最旧的项目，而是删除第二旧的项目。

另一方面，如果您使用 synchronized对于所有方法一致，不需要有线程安全的后端集合:

import java.util.ArrayDeque;

public class CircularBlockingQueue<E> {
    private final ArrayDeque<E> blockingQueue;
    private final int maxSize;

    public CircularBlockingQueue(int size) {
        if(size<1) throw new IllegalArgumentException("size == "+size);
        blockingQueue = new ArrayDeque<>(size);
        maxSize = size;
    }

    public synchronized int size() {
        return blockingQueue.size();
    }

    public synchronized void add(E element) {
        if(blockingQueue.size() == maxSize) {
            blockingQueue.poll();
        }
        blockingQueue.add(element);
        notify();
    }

    public synchronized E take() throws InterruptedException {
        while(blockingQueue.isEmpty()) wait();
        return blockingQueue.remove();
    }
}

但是，如果您可以接受有关最旧元素的较弱保证，则可以使用 BlockingQueue并且不需要任何synchronized :

public class CircularBlockingQueue<E> {
    private final ArrayBlockingQueue<E> blockingQueue;

    public CircularBlockingQueue(int size) {
        blockingQueue = new ArrayBlockingQueue<>(size);
    }

    public int size() {
        return blockingQueue.size();
    }

    public void add(E element) {
        while(!blockingQueue.offer(element)) {
            blockingQueue.poll();
        }
    }

    public E take() throws InterruptedException {
        return blockingQueue.take();
    }
}

必须指出的是，这些解决方案都没有提供“公平性”。因此，如果生产者和消费者线程的数量与队列的容量相比较大，则存在生产者重复删除项目而不重新激活 take() 中阻塞的线程的风险。 。所以你应该始终确保有足够大的容量。