问题标签 [blockingqueue]

I have a class that takes objects from a BlockingQueue and processes them by calling take() in a continuous loop. At some point I know that no more objects will be added to the queue. How do I interrupt the take() method so that it stops blocking?

Here's the class that processes the objects:

And here's the method that uses this class to process objects:

这个问题的标题让我怀疑这是否存在，但仍然：

我对是否有 Java 的BlockingQueue的实现感兴趣，它受大小限制，从不阻塞，而是在尝试将太多元素排入队列时抛出异常。

编辑- 我将 BlockingQueue 传递给一个 Executor，我想它使用它的 add() 方法，而不是 offer()。可以编写一个 BlockingQueue 包装另一个 BlockingQueue 并将 add() 的调用委托给 offer()。

假设我有一个任务是从 java.util.concurrent.BlockingQueue 中提取元素并处理它们。

如果可以动态更改频率，我如何安排/重新安排任务？

• 这个想法是获取数据更新流并将它们批量传播到 GUI
• 用户应该能够改变更新频率

我有一个阻塞的对象队列。

我想写一个线程阻塞直到队列中有一个对象。类似于 BlockingQueue.take() 提供的功能。

但是，由于我不知道我是否能够成功处理该对象，所以我只想 peek() 而不是删除该对象。只有当我能够成功处理它时，我才想删除它。

所以，我想要一个阻塞 peek() 函数。目前，根据 javadocs，如果队列为空，peek() 只会返回。

我错过了什么吗？还有其他方法可以实现此功能吗？

编辑：

如果我只是使用线程安全队列并偷看和睡觉，有什么想法吗？

请注意，我只有一个消费者线程和一个（单独的）生产者线程。我想这不如使用 BlockingQueue 高效......任何评论表示赞赏。

我正在努力寻找实现我的处理管道的最佳方式。

我的生产者将工作提供给 BlockingQueue。在消费者方面，我轮询队列，将我得到的内容包装在一个 Runnable 任务中，然后将其提交给一个 ExecutorService。

这并不理想；当然，ExecutorService 有自己的队列，所以真正发生的是我总是完全排空我的工作队列并填充任务队列，随着任务完成，任务队列会慢慢清空。

我意识到我可以在生产者端排队任务，但我真的不想那样做——我喜欢我的工作队列的间接/隔离是哑字符串；这真的与制片人无关，他们会发生什么。强制生产者对 Runnable 或 Callable 进行排队会破坏抽象，恕我直言。

但我确实希望共享工作队列代表当前的处理状态。如果消费者跟不上，我希望能够阻止生产者。

我很想使用 Executors，但我觉得我在与他们的设计作斗争。我可以喝部分 Kool-ade，还是必须一饮而尽？我在抵制排队任务方面是不是走错路了？（我怀疑我可以设置 ThreadPoolExecutor 以使用 1 任务队列并覆盖它的执行方法来阻止而不是拒绝队列满，但这感觉很糟糕。）

建议？

我有一个BlockingQueue在生产者-消费者情况下使用的实现。我想装饰这个队列，以便记录从中取出的每个对象。我知道简单的实现会是什么样子：只需在构造函数中实现BlockingQueue并接受 a BlockingQueue，所有方法都将委托给它。还有另一种我想念的方式吗？也许是图书馆？有回调接口的东西？

有人知道为什么 java 的 BlockingQueue 没有 putAll 方法吗？这种方法有问题吗？无需完全重新实现 BlockingQueue 即可解决此问题的任何好方法？

我有一些代码，我使用执行器和阻塞队列执行多个任务。结果必须作为迭代器返回，因为这是我工作的应用程序所期望的。但是，任务和添加到队列的结果之间存在 1:N 的关系，因此我不能使用ExecutorCompletionService。在调用 hasNext() 时，我需要知道所有任务何时完成并将所有结果添加到队列中，以便我可以停止从队列中检索结果。请注意，一旦项目被放入队列，另一个线程应该准备好使用（Executor.invokeAll(), 阻塞直到所有任务完成，这不是我想要的，也不是超时）。这是我的第一次尝试，我使用 AtomicInteger 只是为了证明这一点，即使它不起作用。有人可以帮助我了解如何解决这个问题吗？

调用代码如下所示：

我有三个 JavaLinkedBlockingQueue实例，我想take只使用一个线程从它们中读取（操作）。天真的方法是每个队列有一个线程。

有没有像 UNIXselect系统调用那样在 Java 中阻塞队列？

谢谢。

考虑一个BlockingQueue和几个线程等待poll(long, TimeUnit)（可能也在 on take()）。

现在队列是空的，需要通知等待的线程他们可以停止等待。预期的行为是null返回或声明的InterruptedException抛出。

Object.notify()LinkedBlockingQueue由于线程正在等待内部锁，因此无法正常工作。

有什么直截了当的方法吗？