java - 由于发布者和订阅者驻留在受相同资源限制的同一个应用程序中，那么背压如何帮助解决双方的负载问题？

Question

我一直在尝试使用 Java Flow API 构建应用程序。虽然在发布者和订阅者的速度不同的情况下能够在发布者和订阅者之间执行背压的想法，但我不确定它是否真正有帮助，因为发布者和消费者通常驻留在同一个应用程序中；至少，几乎所有在线示例都是如此。

例如，在我的应用程序中，有一个发布者生成从 RabbitMQ 检索的消息，还有一个订阅者处理这些消息。因此，消息在 RabbitMQ 监听器中提交给发布者，如下所示:

@RabbitListener(queues = "rabbit_queue")
public void rabbitHandler(MessageObject message) {
   // do some stuff to the message and then submit it to the publisher
   publisher.submit(message);
}

// Then the message will be processed in the subscriber

如果发布者的生成速度快于订阅者的处理速度，则订阅者可以在 subscription.request(n) 上调用一个较小的 n 值。但是，有两件事我不确定我对 request(n) 如何提供帮助的理解是否正确:

1. 由于在本例中发布者和订阅者都位于同一个 Spring 应用程序中，因此它们几乎共享并受到相同数量的资源的限制。如果订阅者由于发送给它的元素太多而耗尽内存或资源，我们应该能够减少 request(n) 中的 n 值。但这意味着发布者中的缓冲区大小将很快填满。我可以增加发布者中的缓冲区大小，但我也受到订阅者所面临的相同资源量的限制，因为发布者和订阅者都在使用同一组资源的同一应用程序中。那么拥有发布者和 request() 方法所带来的额外复杂性有什么意义呢？
2. 在我看来，发布商通常从某些来源接收其元素。有时，并非所有这些来源都可以受到限制。就我而言，我有一个 RabbitMQ 监听器将消息发送给发布者。但是发布者将这些消息发送到订阅的速率很大程度上取决于rabbitHandler 从 RabbitMQ 队列接收消息的速率。如果 RabbmitMQ 发送消息的速度比发布者的订阅者可以处理的速度快，则消息的缓冲仍然在应用程序内的发布者和订阅者之间完成，并且会出现上述问题。

我很确定我对这个过程的理解有问题，因为对我来说这感觉就像是第 22 条军规的情况。就好像我的两只手只能握住这么多球，而我只是在两只手之间传递球并称之为背压。由于发布者和订阅者都受到相同数量的资源的限制，因为它们都在同一个应用程序中，因此当我可以简单地将消息传递给另一个处理程序并受到相同数量的资源的限制时，具有额外的复杂性有什么好处，如下所示:

public class RabbitMqListener {
    @RabbitListener(queues = "rabbit_queue")
    public void rabbitHandler(MessageObject message) {
       // do some stuff to the message and then submit it to the publisher
       MessageProcessor.process(message);
    }
}

public class MessageProcessor { 
   public static void process(MessageObject message) {
      System.out.println("processing message...");
   }
}

如果有人能帮助我纠正我的理解，那就太好了。

Answer 1

“如果 RabbmitMQ 发送消息的速度比发布者的订阅者可以处理的速度快”

那么您应该尝试将背压反馈扩展到消息的源头，即 RabbmitMQ 发布者。为了这个目标，您可以创建额外的点对点连接。如果您无法减慢 RabbmitMQ 发布者的速度，那么您有 2 个选择:删除一些您无法存储的消息，或者购买性能更高的硬件。