问题标签 [blockingqueue]

我正在编写一个实现队列接口的自定义队列。此实现是线程安全的，并且在某些情况下会阻塞。

普通的 Queue 接口没有提到异常，因此我不能InterruptedException在我的实现中抛出任何异常。

我看到了这个问题的两种解决方案，但它们都不是很令人满意：

1. 移除 Queue 接口并抛出异常。这使得代码无法用于需要队列的外国软件。

2. Throw RuntimeException，这将产生大量非常令人惊讶的软件活动，我不想冒险。

不知何故，诸如ArrayBlockingQueue设法实现Queue和BlockingQueue. 这是要走的路，还是这里的诀窍是什么？

JUnit是否已经有一些用于测试接口的现有测试BlockingQueue？我可以下载一些课程，按下播放键，然后它变成红色（希望是绿色:-)），而我不必花一天时间思考所有测试用例？

我用谷歌搜索但找不到任何东西。

一些想法？

编辑：G_H提供的测试用例非常适合调试基础知识。对于繁重的多线程使用可能还有一个很好的测试，这真的会引发竞争和死锁吗？

我有一个阻塞队列，工作人员从该队列中取出，然后工作大约 1-10 毫秒。

我想弄清楚工人在给定时间范围内活动和空闲的时间的一些相对数字（例如，最后一分钟活动的 70%）。

总结短短的 1ms 周期可能是不准确的，并且获取纳米时间会消耗很多性能。

有没有一些巧妙的方法来完成这样的任务？

我在我的代码中更频繁地使用这些队列工作者设置，因此我希望能够在运行时分析瓶颈。

编辑：我只想知道线程是否可以工作。如果操作系统将线程移出处理焦点，它仍应计数。只有在等待队列锁时才应该算作空闲时间。

Can anyone explain why someone should use the Android Looper feature to create a "pipeline thread" instead of making a normal thread that pulls tasks from a BlockingQueue? On the surface, it seems like two ways to do the same thing.

有什么区别？我什么时候应该使用容量为 1 的SynchronousQueue反对？LinkedBlockingQueue

我目前正在尝试实现一个类以用作通过蓝牙连接传入数据的缓冲区：

我遇到了这个缓冲区的问题。每当我添加一个字节数组时，缓冲区中的所有字节数组都与我刚刚添加的字节数组相同。

我已经尝试在我的其余代码中使用相同类型的阻塞队列（没有使其成为自己的类），它工作正常。问题似乎出在我使用此类时。

我声明类的方式如下：

谁能看到我在哪里犯了错误？

我对此感到非常困惑，并且在过去的三天里我一直在尝试调试。希望有人能够告诉我我做错了什么。

我正在实现一个 BlockingQueue (FIFO) 缓冲区来接收通过蓝牙从我的 PC 流式传输的信息。我正在使用 RealTerm 通过超级终端链接发送预先记录的心电图信号。

我通过添加值然后删除它们来启动应用程序时测试了缓冲区，它似乎可以正常工作。

当我从蓝牙连接接收数据时尝试存储在缓冲区中时，问题就出现了。我不知道我添加的速度是否比 BlockingQueue 可以应付的速度快，但是当我停止数据传输并检查我的缓冲区时，整个缓冲区包含添加的最后一个值。缓冲区的大小是正确的，但内容不正确。

这是我的缓冲区：

我的 BluetoothCommunication 类接收信息并将其发送到缓冲区的部分如下：

所以我的问题的一个例子是：

通过蓝牙连接发送值（从 1 到 20 的 8 位值）。在 IncomingBuffer 类的 insert 方法中，日志消息确认发送了正确的值。当从缓冲区中检索值时，它包含 20 个字节数组，这些数组都包含最后插入的数字 (20)。

关于为什么缓冲区会在其他情况下工作但在蓝牙通信期间不起作用的任何线索？

我有一个情况，我有 2 个阻塞队列。首先我插入一些我执行的任务。当每个任务完成时，它会将一个任务添加到第二个队列中，并在那里执行它们。

所以我的第一个队列很简单：我只是检查以确保它不为空并执行，否则我会中断（）：

第二个我执行以下操作，如您所知，它不起作用：

你将如何解决它，以便第二个 BlockingQueue 不会在 take() 上阻塞，但只有在它知道没有更多项目要添加时才完成。如果第二个线程可以看到第一个阻塞队列，并检查它是否为空并且第二个队列也为空，那么它将中断。

我也可以使用 Poison 对象，但更喜欢别的东西。

注意：这不是确切的代码，只是我在这里写的：

我正在尝试在 pthread、semaphore.h 和 gcc atomic builtins 之上实现一个由循环缓冲区支持的高性能阻塞队列。队列需要处理来自不同线程的多个同时读取器和写入器。

我已经隔离了某种竞争条件，我不确定这是否是对某些原子操作和信号量行为的错误假设，或者我的设计是否存在根本缺陷。

我已将其提取并简化为以下独立示例。我希望这个程序永远不会返回。然而，它确实会在几十万次迭代后返回，并在队列中检测到损坏。

在下面的示例（用于说明）中，它实际上并没有存储任何内容，它只是将一个保存实际数据的单元格设置为 1，将一个空单元格设置为 0。有一个计数信号量（vacancys）表示空单元格的数量，另一个计数信号量（occupants）表示占用单元格的数量。

编写者执行以下操作：

1. 减少职位空缺
2. 原子地获取下一个头索引（mod队列大小）
3. 写信给它
4. 增加居住者

读者反其道而行之：

1. 递减乘员
2. 原子地获取下一个尾索引（mod 队列大小）
3. 从中读取
4. 增加职位空缺

我希望鉴于上述情况，恰好一个线程可以一次读取或写入任何给定的单元格。

任何关于它为什么不起作用或调试策略的想法都值得赞赏。下面的代码和输出...

编译上面的代码如下：

每次输出都不同，但这里有一个示例：

我的系统是 Intel Core 2 上的 Ubuntu 11.10：

谢谢，安德鲁。

我有一个单线程生产者，它创建一些任务对象，然后将它们添加到一个ArrayBlockingQueue（固定大小）中。

我还启动了一个多线程消费者。这是作为固定线程池 ( Executors.newFixedThreadPool(threadCount);) 构建的。然后我将一些 ConsumerWorker 实例提交给这个 threadPool，每个 ConsumerWorker 都有一个对上述 ArrayBlockingQueue 实例的引用。

每个这样的 Worker 都会take()在队列上做一个并处理任务。

我的问题是，让工人知道什么时候没有更多工作要做的最好方法是什么。换句话说，我如何告诉Workers生产者已经完成加入队列，并且从这一点开始，每个worker看到队列为空时应该停止。

我现在得到的是一个设置，我的 Producer 使用回调初始化，当他完成它的工作（向队列中添加东西）时触发该回调。我还保留了我创建并提交到 ThreadPool 的所有 ConsumerWorkers 的列表。当生产者回调告诉我生产者完成时，我可以告诉每个工人。在这一点上，他们应该简单地继续检查队列是否不为空，当它变空时他们应该停止，从而允许我优雅地关闭 ExecutorService 线程池。是这样的

}

这里的问题是我有一个明显的竞争条件，有时生产者会完成，发出信号，而 ConsumerWorkers 会在消耗队列中的所有内容之前停止。

我的问题是同步它的最佳方法是什么，以便一切正常？我是否应该同步检查生产者是否正在运行的整个部分，以及队列是否为空，并在一个块中从队列中取出一些东西（在队列对象上）？我应该只在 ConsumerWorker 实例上同步isRunning布尔值的更新吗？还有什么建议吗？

更新，这是我最终使用的工作实现：

}

这在很大程度上受到了 JohnVint 在下面的回答的启发，只做了一些小的修改。

=== 由于@vendhan 的评论而更新。

感谢您的关注。你是对的，这个问题中的第一段代码（在其他问题中）有一个while(isRunning || !inputQueue.isEmpty())没有真正意义的代码段。

在我的实际最终实现中，我做了一些更接近您替换“||”的建议 (or) with "&&" (and)，意思是每个工人（消费者）现在只检查他从列表中得到的元素是否是毒丸，如果是的话就停止（所以理论上我们可以说工人有正在运行且队列不能为空）。