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

Question

我想用Java实现各种各样的发布者/订阅者模式，但目前已经没有主意了。

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

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

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

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

有什么更好的方法吗？我认为这应该是一种很常见的模式。

Answer 1

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

我会改变这个建筑。我最初考虑了每个用户的队列，但我不喜欢这种机制。例如，如果第一个订阅者需要更长的时间来运行，那么所有的作业都将在该队列中结束，您将只执行一个线程的工作。

由于您必须按顺序运行订阅者，因此我有一个线程池，通过所有订阅者运行每条消息。对订阅者的调用需要是可重入的，这可能是不可能的。

因此，您将有一个包含 10 个线程的池（假设），每个线程都从发布者的队列中取消排队，并执行如下操作：

public void run() {
    while (!shutdown && !Thread.currentThread().isInterrupted()) {
        Article article = publisherQueue.take();
        for (Subscriber subscriber : subscriberList) {
           subscriber.process(article);
        }
    }
}

Answer 2

毫无疑问

每个订阅者都有自己的阻塞队列，发布者将对象放入每个订阅者的队列

这是要走的路。您可以使用线程方法将其放入队列…所以如果一个队列已满，发布者不会等待…

s1 s2 s3是订阅者，addToQueue是每个订阅者中添加到腐蚀队列的方法。addQueue方法是等待队列非空…所以调用addQueue将是阻塞调用理想的同步code…

然后，在publisher中，您可以执行类似以下代码的操作

注意：代码可能未处于正常工作状态。但应该给你想法。

List<subscriber> slist;// Assume its initialised
public void publish(final String message){

    for (final subscriber s: slist){


          Thread t=new Thread(new Runnable(){
             public void run(){
                s.addToQueue(message);
             }
           });

      t.start();
    }

}

Answer 3

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