阻塞队列设计

n个线程产生到BlockingQueue。当队列已满时，消费者会排空队列并执行一些处理。阻塞队列设计

我应该如何决定以下两种实施方式？

选择A：消费者定期轮询队列以检查它是否已满，所有写入者正在等待（毕竟这是一个阻塞队列：）。

选择B：我使用同步的“put”方法实现自己的队列。在放置提供的元素之前，我测试队列是否接近满（全部减1元素）。然后，我把这个元素，并通知我的消费者（正在等待）。

第一种解决方案是最简单的方法，但可以进行轮询;这让我很烦恼。第二种解决方案在我看来更容易出错，更多需要更多编码。

来源

2012-05-15 charly' s

您还可以使用其他方式来沟通“开始处理”，例如， CountdownLatch。此外，为什么消费者不能只获取所有元素，并决定何时自行开始处理它们（例如在拥有N个元素之后？） –

然后，我的消费者会消耗所有元素，并在达到特定数量时元素，他处理它们。在这种情况下，他会一直工作，并试图达到极限我喜欢那个:)我会试试看看它是如何工作的... –

第二种解决方案显然更好。它并不复杂。您可以继承或包装任何其他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; }

来源

2012-05-15 10:40:00 AlexR

我想到了那个，但是在返回false的第一个offer和我的第二个“offer”之间，我可能会再次被卡住，因为有其他线程可能会填满我的队列。 –

我不知道有没有这样的实现需要队列：消费者是等待，而队列已满，只有当它完全排出，并开始处理。

您的排队应该被阻止消费者，直到它变满。认为您需要重写drain（）方法，使其在队列变满时等待。比你的消费者只是打电话等排水法。不需要从生产者到消费者的通知。

来源

2012-05-15 11:10:27 alexey28

使用观察者模式。让您的消费者注册队列通知器。当一个生产者执行放置时，队列将决定是否通知任何监听者。

来源

2012-05-15 11:17:02

我会建议编写你的代理队列，它将在内部与一个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 
     } 
    } 
}

来源

2012-05-15 11:53:58 Drona

你是否建议使用Exceptions来控制流程？这通常是一个糟糕的主意，并且令人不悦。 – Brady

代码仅仅是指示性的，不打算提供实现，只是为了给出一个想法。但是，你说得对！ – Drona

我喜欢这个有2个队列的选项，我最近在C++嵌入式环境中使用它，它真的减少了生产者和消费者之间的锁定争用。在添加到队列中时，我做了一个简单的检查，而不是例外情况。 – Brady

我用了CountdownLatch，它很简单，效果很好。感谢您的其他想法:)

来源

2012-05-16 08:10:22

阻塞队列设计

回答

相关问题