java多线程: How to know when a consumer thread should remove an element from queue

Question

我有一个生产者消费者模型，arduino 生成数据包，在我的电脑内我有一个 java 程序，它获取这些数据包并将它们放入 BlockingQueue 中。现在有线程处理这些数据包。

假设 X 是生产者线程，A、B 和 C 是消费者线程。有一个队列，所有消费者都可以引用它。消息(数据包)是不可变的对象(即消费者无法更改元素的状态)。我的问题是我如何知道所有线程都已使用队列内的特定元素完成，以便我可以将其删除？

这是我的消费者 run() 方法:

@Override
public void run()
{
    while (isStarted && !queue.isEmpty()) {
            updateMap(queue.peek());
    }
}

我正在考虑的一种设计是使用有界队列。当生产者发现队列已满时，它会删除第一个元素。但我不确定这是否是一种安全的方法。我读过this教程和其他一些内容，我得到的是:

Producer should wait if Queue or bucket is full and Consumer should wait if queue or bucket is empty.

如果这听起来很明显，我很抱歉，但我是多线程编程的新手，而且代码的并发性对我来说听起来很可怕。

编辑:

所有A、B 和C 都独立。一个用于统计，一个用于更新网络 map 等。

编辑2:

正如 @Augusto 所建议的，还有另一种方法，其中 A、B 和 C 都有自己的队列。我的网络监听器将数据包传递到每个队列并由它们处理。它可以工作，但是如何仅使用一个队列来完成此操作？或是否可以仅使用一个队列来实现此场景？以及如果答案是肯定的。 我需要如何以及何时从队列中删除元素？

Answer 1

(编辑后我的评论的后续内容)我建议每个线程有一个不同的队列。这实际上是一个众所周知的模式，称为 publish-subscribe .

取自上面的链接:

A Publish-Subscribe Channel works like this: It has one input channel that splits into multiple output channels, one for each subscriber. When an event is published into the channel, the Publish-Subscribe Channel delivers a copy of the message to each of the output channels. Each output channel has only one subscriber, which is only allowed to consume a message once. In this way, each subscriber only gets the message once and consumed copies disappear from their channels.

ConcurrentLinkedQueue 和 BlockingQueue 之间的主要区别在于，您可以添加对 BlockingQueue 上元素数量的限制。如果从网络读取的生产者生成数据的速度比消费者处理数据的速度快，那么这很好。如果您使用无界队列并且这种情况持续一段时间，您最终会遇到 OutOfMemoryError。