问题标签 [blockingqueue]

从文件的内容填充队列时，深度似乎永远不会增加，因为在此实现中没有添加元素。

替换offer为 时add，如果抛出异常

请问我做错了什么？为什么插入第一个元素后队列立即满了？

我有一个要求，我需要读取文本文件然后对其进行转换并将其写入其他文件。我希望以并行方式执行此操作，例如一个线程用于读取，一个用于转换，另一个用于写入。
现在要在线程之间共享数据，我需要一些通道，我正在考虑使用BlockingQueue它，但如果可用的话，我想探索一些其他（更好的）替代方案。
Guava 有一个EventBus但不确定这是否适合该要求。从性能的角度来看，还有哪些其他替代方案可用，以及哪一个是最好的。

在 java 中，有一个很好的包 java.util.concurrent 包含 BlockingQueue 接口的实现。

我在 Haskell 中需要类似的东西，所以它能够

• 在内存中保持固定大小的队列
• 队列为空时阻止读取操作（获取）
• 提供时间框块，如果队列为空且超时，则将返回 Nothing
• 类似于 put 操作 - 阻塞直到队列具有时间盒版本的容量

可能这可以通过 STM 或阻塞事务来实现 - 但我无法在 hackage 上找到类似的东西。

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

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

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

更新

也许要澄清一下：

情况总是如下。可以有多个生产者将项目添加到队列中，但从队列中获取项目的消费者永远不会超过一个。

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

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

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

我有一个使用阻塞的生产者/消费者的实现，我像这样启动它们

Producer 内部是这样的

消费者看起来像这样

但是一段时间后它变得非常慢，无论如何我可以保持它非常快吗？

我有以下阻塞队列；

在哪里

我有一个将消息放入队列的生产者。

现在我的目标是创建能够从队列中获取特定作者的特定消息的消费者？是否可以？

如果不是 BlockingQueue 的替代品是什么？

我最近看到了 BlockingQueue 的以下入队实现（来源）

为什么while循环是必要的，并且可以while替换为if (this.queue.size() == this.limit)

似乎方法入队是同步的，因此一次只能在方法主体中执行 1 个线程并调用wait(). 一旦线程被通知，它就不能继续前进而不this.queue.size() == this.limit再次检查条件吗？

我有几个使用 ArrayBlockingQueue 的工人。

每个工作人员从队列中取出一个对象，对其进行处理，结果可以获得多个对象，这些对象将被放入队列中以进行进一步处理。所以，工人=生产者+消费者。

工人：

主要的：

当没有更多工作时，停止工人的最佳方法是什么？

首先，我的想法是定期检查队列中有多少项目以及当前正在处理的项目数量。如果两者都为零，则在 ExecutorService 上执行类似“shutdownNow()”的操作。但我不确定这是最好的方法。

我已经编写了以下代码，但我觉得我在某个地方出错了：

这里 add(String) 将字符串添加到队列中。每当它接收到来自 UDP 端口的输入时，就会调用它。process() 处理队列并将其发送给 Message 类进行处理。输出 Locked and Released Print Statements 的顺序不理想。

编辑

我的预期答案应该是：如果它在 Producer 中添加，则由 Producer 锁定 -> 然后添加到队列 -> 锁定释放。同样的方式将在消费者中。但是操作不应该交错，即一旦被生产者锁定打印它不应该打印被消费者锁定然后释放锁。

我想通过使用 ExecutorService 和 ArrayBlockingQueue 了解我对生产者消费者设计的理解是否正确。我知道有不同的方法来实现这个设计，但我想，最后，这取决于问题本身。

我不得不面对的问题是：我有一个从一个大文件（6GB）读取的 ONE 生产者；它逐行读取并将每一行转换为一个对象。它将对象放在 ArrayBlockingQueue 中。

消费者（少数）从 ArrayBlockingQueue 中获取对象并将其保存到数据库中。

现在，显然生产者比消费者快得多；将每一行转换为一个对象需要几分之一秒，但对于消费者来说需要更长的时间。

所以......如果我希望通过这样做来加速这个过程：我创建了 2 个分类的“ProducerThread”和“ConsumerThread”，它们共享 ArrayBlockingQueue。两者之间协调的线程如下所示：

我的问题是：

1. 上面的设计实际上会同时使用每个线程吗？我的电脑是两个 2.4GHz 四核。

2. 我不确定 Future 和 .get() 的用途是什么？

顺便说一下，结果很快（考虑到第一个版本是连续的，需要 3 小时）现在大约需要 40 分钟（但可能还有改进的余地）。

感谢任何指针