生产者/消费者 - 生产者将数据添加到集合而不阻塞，消费者批量消费来自集合的数据

Producer/Consumer - producer adds data to collection without blocking, consumer consumes data from collection in batch(生产者/消费者 - 生产者将数据添加到集合而不阻塞，消费者批量消费来自集合的数据)

问题描述

我有一个有点不寻常的生产者/消费者用例.我有一些生产者的真实用例，我希望他们能够将对象添加到集合中而不会阻塞.消费者(只有一个)应该阻塞，直到集合中有一定数量的对象可用(例如 500)，然后批量消费它们.虽然少于 500 个，但它应该阻塞并等待集合填满.我不介意队列是否在短时间内超过此值(700、1000 等).

I have a Producer/Consumer usecase which is a bit unusual. I have a real world use case with some producers which I want them to be able to add objects into a collection without blocking. The consumer (just one) should block until a certain amount of objects are available in the collection (eg. 500) and then consume them in bulk. While there are less than 500 it should block and wait for the collection to fill. I don't mind if the queue exceeds this value (700, 1000 etc.) for short amount of times.

我目前似乎没有找到解决这个确切问题的解决方案.我正在考虑使用 ConcurrentLinkedQueue 并让消费者定期检查队列是否有足够的数据，但这似乎适得其反.

I currently don't seem to find a solution to fix this exact problem. I was thinking about using a ConcurrentLinkedQueue and to have the consumer periodically check if the queue has enough data, but that seems rather counterproductive.

另一个想法是使用 LinkedBlockingQueue.生产者不会阻塞(除非队列已满，这意味着它具有 Integer.MAX_VALUE 值 - 这不是我的情况，所以一切都很好).消费者将执行 queue.take() 并将元素添加到内部集合中.当内部集合达到 500 个元素时，它将批量消耗它们.

Another idea is to use a LinkedBlockingQueue. The producers are not going to block (unless the queue is full which means it has Integer.MAX_VALUE values - this is not my case, so it's all good). The consumer will do a queue.take() and add the elements into an internal collection. When the internal collection reaches 500 elements it is going to consume them in batch.

你有什么建议吗?

谢谢！

推荐答案

最后我选择了 LinkedBlockingQueue.

In the end I went with the LinkedBlockingQueue.

生产者将项目添加到队列中.

The producers add items into the queue.

消费者将执行 queue.poll() 并将项目保存到内部集合中.当内部集合达到 500 个元素时，它将批量消耗它们.我还可以设置一些超时.例如，如果 X 秒过去了，我将使用该集合，即使它的项目少于所需的项目(例如 220).

The consumer will do a queue.poll() and keep the items into an internal collection. When the internal collection reaches 500 elements it is going to consume them in bulk. I can also set some timeouts. For instance, if X seconds has passed I'll consume the collection even if it has less items than required (eg. 220).

这篇关于生产者/消费者 - 生产者将数据添加到集合而不阻塞，消费者批量消费来自集合的数据的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持编程学习网！