java - 发布者/订阅者模式的并发实现

Question

我想使用 Java 实现各种发布者/订阅者模式，但目前没有想法。

有 1 个发布者和 N 个订阅者，发布者发布对象，然后每个订阅者需要按照正确的顺序处理每个对象一次且仅一次。发布者和每个订阅者在自己的线程中运行。

在我最初的实现中，每个订阅者都有自己的阻塞队列，发布者将对象放入每个订阅者的队列中。这工作正常，但如果任何订阅者的队列已满，发布者将被阻止。这会导致性能下降，因为每个订阅者处理对象所需的时间不同。

然后在另一个实现中，发布者将对象保存在自己的队列中。除了该对象之外，还有一个 AtomicInteger 计数器与该对象以及订阅者的数量相关联。然后，每个订阅者查看队列并减少计数器，并在计数器达到零时将其从队列中删除。

通过这种方式，发布者不会阻塞，但现在订阅者需要等待彼此处理该对象，从队列中删除该对象，然后才能查看下一个对象。

有没有更好的方法来做到这一点？我认为这应该是一种很常见的模式。

Answer 1

您的“多队列”实现是正确的选择。我认为您不必担心一个完整的队列阻塞了生产者，因为完成的总体时间不会受到影响。假设您有 3 个消费者，其中两个以每秒 1 个的速度消费，第三个以每五秒 1 个的速度消费，同时生产者以每两秒 1 个的速度生产。最终第三个队列将填满，因此生产者将阻塞它，并且也将停止将项目放入第一个和第二个队列中。有很多方法可以解决这个问题，但它们不会改变第三个消费者将始终成为瓶颈的事实。如果您正在生产/消费 100 个项目，那么由于第三个消费者(5 秒乘以 100 个项目)，这将至少需要 500 秒，即使第一个和第二个消费者在 200 秒后完成，情况也会如此(因为你做了一些聪明的事情，允许生产者继续填充他们的队列，即使在第三个队列已满之后)或者如果他们在 500 秒后完成(因为生产者在第三个队列上阻塞)。