Есть ли (неограниченная) справедливая очередь блокировки в Java?

Question

Существует ли реализация блокировки очереди, которая гарантирует справедливую операцию take (), если несколько потребителей удаляют элемент из одной очереди.Я проверил LinkedBlockingQueue, LinkedTransferQueue и похоже, что оба они несправедливы.ArrayBlockingQueue обеспечивает честную работу, но ограниченную.

Answer 1

Мы можем реализовать неограниченную очередь честного блокирования, используя неограниченную очередь, такую ​​как очередь ConcurrentLinked и честный семафор.Приведенный ниже класс реализует не все методы интерфейса BlockingQueue, а лишь некоторые из них для демонстрационных целей.Метод main () записывается только в качестве теста.

public class FairBlockingQueue<T> {

    private final Queue<T> queue;
    private final Semaphore takeSemaphore;

    public FairBlockingQueue() {
        queue = new ConcurrentLinkedQueue<T>();
        takeSemaphore = new Semaphore(0, true);
    }

    public FairBlockingQueue(Collection<T> c) {
        queue = new ConcurrentLinkedQueue<T>(c);
        takeSemaphore = new Semaphore(c.size(), true);
    }

    public T poll() {
        if (!takeSemaphore.tryAcquire()) {
            return null;
        }
        return queue.poll();
    }

    public T poll(long millis) throws InterruptedException {
        if (!takeSemaphore.tryAcquire(millis, TimeUnit.MILLISECONDS)) {
            return null;
        }
        return queue.poll();
    }

    public T take() throws InterruptedException {
        takeSemaphore.acquire();
        return queue.poll();
    }

    public void add(T t) {
        queue.add(t);
        takeSemaphore.release();
    }

    public static void main(String[] args) throws Exception {
        FairBlockingQueue<Object> q = new FairBlockingQueue<Object>();
        Object o = q.poll();
        assert o == null;
        o = q.poll(1000);
        assert o == null;

        q.add(new Object());
        q.add(new Object());
        q.add(new Object());

        o = q.take();
        assert o != null;
        o = q.poll();
        assert o != null;
        o = q.poll(1000);
        assert o != null;

        o = q.poll();
        assert o == null;
    }
}

Answer 2

Политика справедливости может быть указана для SynchronousQueue :

очередь, построенная с набором справедливости к истине предоставляет доступ к потокам в FIFO заказ

Answer 3

PriorityBlockingQueue