阻塞队列设计

Question

N个线程生成一个BlockingQueue。当队列已满时，消费者会排空队列并进行一些处理。

我应该如何在以下两种实现方式之间做出选择？

选项A:消费者定期轮询队列以检查它是否已满，所有写入器都在等待(毕竟这是一个阻塞队列：)。

选择B:我用一个同步的"put“方法实现我自己的队列。在放置所提供的元素之前，我测试队列是否接近满(满减去1个元素)。然后我放入元素并通知我的消费者(它正在等待)。

第一个解决方案是最简单的，但是会进行轮询；这让我很恼火。在我看来，第二种解决方案更容易出错，而且需要更多的编码。

Answer 1

我建议编写代理队列，它将在内部包装一个queue实例和一个Exchanger实例。您的代理方法将调用委托给内部队列。添加时检查内部队列是否已满，当内部队列已满时，与使用者线程交换内部队列。使用者线程将交换空队列，以换取已填满的队列。您的代理队列将继续填充空队列，而使用者可以继续处理已填充的队列。这两个活动可以并行运行。当双方都准备好时，他们可以再次交换。

class MyQueue implements BlockingQueue{
    Queue internalQueue = ...
    Exchanger<Queue> exchanger;

    MyQueue(Exchanger<BlockingQueue> ex){
    this.exchanger = ex;
    }

     .
     .
     .

    boolean add (E e) {
      try{
        internalQueue.add(e);
      }catch(IllegalStateException ise){
        internalQueue = exchanger.exchange(internalQueue);
      }
      internalQueue.add(e);     
    }

}

class Consumer implements Runnable {
    public void run() {
        Queue currentQueue = new empty queue;
        while (...){
           Object o = currentQueue.remove();
           if (o == null){
              currentQueue = exchanger.exchange(currentQueue);
              continue;
           }
           //cast and process the element
        } 
    }
}

Answer 2

第二种解决方案显然更好。而且它也不是那么复杂。您可以继承或包装任何其他BlockingQueue并覆盖它的方法offer()，如下所示:调用“真正的”offer()。如果返回true，则退出。否则触发工作线程工作，并立即调用超时的offer()。

下面是几乎是伪代码：

public boolean offer(E e) {
    if (queue.offer(e)) {
        return true;
    }
    boolean result = queue.offer(e, timeout, unit); // e.g. 20 sec. - enough for worker to dequeue at least one task from the queue, so the place will be available.
    worker.doYouJob();
    return result; }

Answer 3

我不知道是否有您需要的队列实现:当队列变满时，消费者正在等待，并且只有当它完全耗尽并开始处理时。

您的队列应该被阻塞，直到它变满为止。我认为您需要重写drain()方法，使其在队列变满时等待。而你的消费者只需调用并等待排出方法。不需要从生产者到消费者的通知。