带有批处理的 Java BlockingQueue？

Question

我对与 Java BlockingQueue 相同的数据结构很感兴趣，除了它必须能够批处理队列中的对象。换句话说，我希望生产者能够将对象放入队列中，但让消费者在 take() 上阻塞，直到队列达到一定大小(批处理大小)。

然后，一旦队列达到批量大小，生产者必须在 put() 上阻塞，直到消费者消耗完队列中的所有元素(在这种情况下，生产者将开始生产消费者再次阻塞，直到再次到达该批处理。

是否存在类似的数据结构？或者我应该写它(我不介意)，我只是不想在外面有东西的时候浪费我的时间。

---

更新

也许可以澄清一下:

情况总会是这样的。可以有多个生产者向队列中添加项目，但永远不会有超过一个消费者从队列中取出项目。

现在，问题是有多个并行和串行的设置。换句话说，生产者为多个队列生产元素，而消费者本身也可以是生产者。这可以更容易地被认为是生产者、消费者-生产者和最终消费者的有向图。

生产者应该阻塞直到队列为空的原因 (@Peter Lawrey) 是因为每个队列都将在一个线程中运行。如果您让它们在空间可用时简单地生产，您最终会遇到这样一种情况，即您有太多线程试图同时处理太多事情。

也许将其与执行服务结合可以解决问题？

Answer 1

我建议您使用 BlockingQueue.drainTo(Collection, int) .您可以将它与 take() 一起使用，以确保获得最少数量的元素。

使用这种方法的优点是您的批量大小会随着工作负载动态增长，并且生产者不必在消费者忙碌时阻塞。即它会针对延迟和吞吐量进行 self 优化。

---

要完全按照要求实现(我认为这是个坏主意)，您可以使用带有繁忙消费线程的 SynchronousQueue。

即消费线程做一个

 list.clear();
 while(list.size() < required) list.add(queue.take());
 // process list.

只要消费者忙，生产者就会阻塞。