生产者/消费者:任何生产者/消费者不得阻止其他生产者/消费者

Question

我正在设计一个系统，其中将有n生产者和m消费者，其中n和m是数字，n != m。

我想把系统设计成这样，

1. 任何生产者在生产时都不应阻止其他生产者。
2. 任何消费者在消费时都不应阻止其他消费者。
3. 生产者和消费者在生产/消费时互相阻碍

在java中，如果我使用同步关键字，那么它将阻塞相应的调用方。

我不知道我应该用什么数据结构和算法来实现这个系统。

有人能在这方面给我提供帮助/指点吗？

Answer 1

你可能想要像ConcurrentLinkedQueue这样的东西。这样做的目的是创建一个队列。您的n个生产者将工作项添加到队列中，并且m个使用者从队列中读取工作项。制片人只是：

while not done
    create work item
    add work item to queue

消费者也同样简单：

while not done
    get next work item from queue
    process work item

ConcurrentLinkedQueue方法处理添加和删除项，并根据需要与其他生产者和使用者同步。

唯一真正的缺点是，您必须轮询队列，以查看是否有项目。因此，您可能需要一个自动重置事件，每当将项添加到队列中时，该事件就会被触发。例如：

add work item to queue
set ItemAvailable event

使用者将轮询队列，如果没有可用的项，则等待事件：

while not done
    while ((item = queue.poll) == null)
        wait on ItemAvailable event
    process item

看看我所链接的例子。使用起来并不难。

Answer 2

取决于您需要做多少繁重工作，以及您的解决方案需要扩展得有多好，RxJava有一些陡峭的学习曲线，但是一旦您通过它，它可能是最优雅的、可伸缩的和执行的解决方案。

在不同的线程中运行所有生产者，将它们与Merg()组合起来，将使用者移动到使用.observeOn(Scheduler.newThread())的未绑定缓冲区上的线程。

如果您需要在多个系统上并行运行的东西，请查看mapreduce。

如果您需要在光谱的另一端(一些简单的东西)，只需坚持一个ConcurrentQueue。这不支持多播，但至少解决了问题的生产者一方。

Answer 3

您需要一种方法，使每个操作都是原子的和不可中断的，所以是的，在我看来，最好的方法是在方法上使用同步修饰符来设置锁。

另一个有趣的方法是使用原子变量-> http://baptiste-wicht.com/posts/2010/09/java-concurrency-atomic-variables.html

这取决于您在这些生产者/消费者结构中的数据。