了解背压

流由发布者和订阅者的链组成。发布者产生新的价值，而订阅者使用它们。在这两者之间放置一个缓冲区，该缓冲区可以保存值，直到使用者准备使用它们为止。下图说明了单个发布者和订阅者的链：

同步发布者和订阅者按顺序运行

如果发布者和订阅者同时操作，则发布者可能会使缓冲区溢出。当发布者发布商品的速度快于订阅者消费商品的速度时，就会发生这种情况。限制发布者以避免订阅者溢出就是我们所说的背压。

如果发布者和订阅者是同步的；当缓冲区即将溢出时，我们可以阻止生产者。Java标准库附带了一个Queue实现此称为阻塞队列。此队列将阻止在完整队列上执行放置操作的线程，并在从队列中取出项目后取消阻止生产者。

注意封锁

放置操作阻塞意味着什么？这意味着将在其上运行该操作的线程被驻留，从而释放正在其上运行的CPU。操作系统将安排一个或多个线程同时运行。当订户从队列中取出一个元素时，生产者的线程将被释放，生产者的执行将继续。

通过停放生产者的线程，阻塞队列使生产者无法超过订阅者。因为直到队列中有更多可用空间，生产者才被授予CPU的时间。

在单独的线程上运行的同步发布者和订阅者

同步流可以阻塞生产者和订阅者在其上执行的线程，因此它们可以依赖BlockingQueue来保证背压。但是，如果我们想实现一个异步运行的流，那么不允许这些流阻塞生产者和订阅者执行的线程。因为它们是非阻塞的，所以它们不能依赖BlockingQueue来防止缓冲区溢出。

异步发布者和订阅者在单独的线程上

因为异步代码不能依赖BlockingQueue，所以需要通过异步并发原语来模拟流控制。在Java运行时中，唯一可用于编写异步代码的原语是回调。我们如何使用回调来限制生产者？

注意异步通信原语和promises（承诺）

期货或promises也依赖于回调，你可以通过提供一个回调来创建它们，一旦操作完成，该回调将被调用。期货有助于组成多个异步任务，但在其下没有其他并发原语。

模拟流量控制

让我们看一下响应流规范的一部分，以了解它如何使用回调来模拟流控制。在响应流中，发布者在订阅者上使用回调来发送元素：

反应流订阅者接口的简化变体

由于订户可以与生产者同时进行操作，因此单个回调不能保证流控制。因此，响应流引入了第二个回调，而不是从发布者到订阅者，而是从订阅者到发布者。此回调是订阅者对生产者的订阅的一部分。在此订阅上，通过允许订阅者传达其可以处理的项目数来模拟背压：

反应流订阅接口的简化变体

基于回调的发布者和订阅者

第二个回调允许响应流在不阻塞底层线程的情况下模拟流控制。在阻塞实现保证发布者不会溢出订阅者的情况下（因为操作系统可以驻留生产者的线程），非阻塞变量没有这种保证。发布者不必遵守订阅者通过请求回调提供的限制。

对运营商的影响

运营商（例如地图）既是订户，因为它从上游接收项目，又是发布者，因为它为下游生产项目。运算符用于通过对流通过的项目做出反应来更改流。由于操作员是发布者，因此需要考虑其实施过程中来自下游的背压。

在操作员的实施中考虑背压意味着操作员的实施者有责任遵守流量控制合同。不允许向下游发送比请求更多的项目。并且运营商需要确保从上游请求足够的物品。为了不使溪流停滞

为了演示流如何停止，我们可以从Rx文档中查看filter运算符。由于筛选器运算符不会发出其接收的所有元素，因此需要为每个忽略的元素从上游请求其他项：

过滤器运算符的简化实现

实现缓冲区运算符时，解决异步流中的背压所需的同步变得更加明显。在这里，排水方法模拟了流量控制以及取消，错误处理和完成的问题。同步流的BlockingQueue将提供此模式而无需其他同步。异步变体需要复杂的编排。

结论

流需要流控制，以防止发布者溢出订阅者。同步流可以依靠诸如BlockingQueue之类的原语来提供此流控制，而异步流则需要依赖于回调来模拟背压。这在发布者和订阅者之间创建了一个复杂的合同，这使得异步流的实现和使用比同步流更加困难，并且更容易出错。这并不是说异步流是不好的。但这是为了证明，尽管有用，但异步流是建立在复杂抽象之上的，即通过回调模拟流控制。如果我们可以在异步阻塞队列之上构建异步流，那将结合两个方面的优势。