Java BlockingQueue与批处理？

Question

我对与Java BlockingQueue相同的数据结构感兴趣,但它必须能够批处理队列中的对象.换句话说,我希望生产者能够将对象放入队列,但是使用消费者块,take()直到队列达到一定的大小(批量大小).

然后,一旦队列达到批量大小,生产者必须阻止,put()直到消费者已经消耗了队列中的所有元素(在这种情况下,生产者将再次开始生产并且消费者块直到再次到达批次).

是否存在类似的数据结构？或者我应该写它(我不介意),如果有什么东西,我只是不想浪费我的时间.

---

UPDATE

也许稍微澄清一下事情:

情况总是如下.可以有多个生产者向队列添加项目,但永远不会有多个消费者从队列中获取项目.

现在,问题是这些设置有多个并行和串行.换句话说,生产者为多个队列生产物品,而消费者本身也可以是生产者.这可以更容易地被视为生产者,消费者 - 生产者和最终消费者的有向图.

生产者应该阻塞直到队列为空(@Peter Lawrey)的原因是因为每个都将在一个线程中运行.如果你让它们只是在空间可用的情况下生成,你最终会遇到太多线程试图同时处理太多东西的情况.

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

Answer 1

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

使用此方法的优点是您的批量大小随工作负载动态增长,生产者不必在消费者忙时阻止.即它自我优化延迟和吞吐量.

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

即消费线程做了

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

生产者将在消费者忙碌时阻止.