上一讲，我们通过分析传统开发模式和响应式编程实现方法之间的差别，引出了数据流的概念。我们知道响应式系统都是通过对数据流中每个事件进行处理，来提高系统的即时响应性的。

那么今天这一讲，我们就先从“流”的概念出发，并引入响应式流程规范，从而分析响应式编程中所包含的各个核心组件。希望通过这一讲的学习，你可以掌握通过背压机制对流量进行高效控制的系统方法，并基于响应式流规范来实现背压机制。

流的概念

简单来讲，所谓的流就是由生产者生产并由一个或多个消费者消费的元素序列。这种生产者/消费者模型也可以被称为发布者/订阅者模型，我在上一讲中已经介绍过这个模型。而关于流的介绍，我将从两方面入手，首先明确流的分类，然后再来讨论如何进行流量控制，流量控制是讨论数据流的核心话题。

流的处理模型

关于流的处理，存在两种基本的实现机制。一种就是传统开发模式下的“拉”模式，即消费者主动从生产者拉取元素；而另一种就是上一讲中分析的“推”模式，在这种模式下，生产者将元素推送给消费者。相较于“拉”模式，“推”模式下的数据处理的资源利用率更好，下图所示的就是一种典型的推模式处理流程。

图 1 推模式下的数据流处理方式示意图

上图中，数据流的生产者会持续地生成数据并推送给消费者。这里就引出了流量控制问题，即如果数据的生产者和消费者处理数据的速度是不一致的，我们应该如何确保系统的稳定性呢？

流量控制

先来看第一种场景，即生产者生产数据的速率小于消费者的场景。在这种情况下，因为消费者消费数据没有任何压力，也就不需要进行流量的控制。

现实中，更多的是生产者生产数据的速率大于消费者消费数据的场景。这种情况比较复杂，因为消费者可能因为无法处理过多的数据而发生崩溃。针对这种情况的一种常见解决方案是在生产者和消费者之间添加一种类似于消息队列的机制。我们知道队列具有存储并转发的功能，所以可以由它来进行一定的流量控制，效果如下图所示。

图 2 添加队列机制之后的生产者/消费者场景示意图

现在，问题的关键就转变为如何设计一种合适的队列。通常，我们可以选择三种不同类型的队列来分别支持不同的功能特性。

无界队列

第一种最容易想到的队列就是无界队列（Unbounded Queue），这种队列原则上拥有无限大小的容量，可以存放所有生产者所生产的消息，如下图所示。

图 3 无界队列结构示意图

显然，无界队列的优势就是确保了所有消息都能得到消费，但显然会降低系统的回弹性，因为没有一个系统拥有无限的资源。一旦内存等资源被耗尽，系统可能就崩溃了。

有界丢弃队列

与无界队列相对的，更合适的方案是选择一种有界队列。为了避免内存溢出，我们可以使用这样一个队列，一般队列的容量满了，就忽略后续传入的消息，如下图所示。

图 4 有界丢弃队列结构示意图

上图中，可以看出这个有界队列的容量为 6，所以第 7 和第 8 个元素被丢弃了。然后当消费者消费了一部分消息之后，队列出现了新的空闲位置，后续的消息就又被填充到队列中。当然，这里可以设置一些丢弃元素的策略，比方说按照优先级或先进先出等。

有界丢弃队列考虑了资源的限制，比较适合用于允许丢消息的业务场景，但在消息重要性很高的场景显然不可能采取这种队列。

有界阻塞队列

如果需要确保消息不丢失，则需要引入有界阻塞队列。在这种队列中，我们会在队列消息数量达到上限后阻塞生产者，而不是直接丢弃消息，如下图所示。

图 5 有界阻塞队列结构示意图

上图中，队列的容量同样是 6，所以当第 7 个元素到来时，发现队列已经满了，那么生产者就会一直等到队列空间的释放而产生阻塞行为。显然，这种阻塞行为是不可能实现异步操作的，所以结合上一讲中的讨论结果，无论从回弹性、弹性还是即时响应性出发，有界阻塞队列都不是我们想要的解决方案。

背压机制

讲到这里，我们已经明确，纯“推”模式下的数据流量会有很多不可控制的因素，并不能直接应用，而是需要在“推”模式和“拉”模式之间考虑一定的平衡性，从而优雅地实现流量控制。这就需要引出响应式系统中非常重要的一个概念——背压机制（Backpressure）。

什么是背压？简单来说就是下游能够向上游反馈流量请求的机制。通过前面的分析，我们知道如果消费者消费数据的速度赶不上生产者生产数据的速度时，它就会持续消耗系统的资源，直到这些资源被消耗殆尽。

这个时候，就需要有一种机制使得消费者可以根据自身当前的处理能力通知生产者来调整生产数据的速度，这种机制就是背压。采用背压机制，消费者会根据自身的处理能力来请求数据，而生产者也会根据消费者的能力来生产数据，从而在两者之间达成一种动态的平衡，确保系统的即时响应性。

响应式流规范

关于流量控制我们已经讨论了很多，而针对流量控制的解决方案以及背压机制都包含在响应式流规范中，其中包含了响应式编程的各个核心组件，让我们一起来看一下。

响应式流的核心接口

在 Java 的世界中，响应式流规范只定义了四个核心接口，即 Publisher 、Subscriber 、Subscription 和 Processor<T,R>。

Publisher

Publisher 代表的就是一种可以生产无限数据的发布者，该接口定义如下所示。

1
2
3

public interface Publisher<T> {
    public void subscribe(Subscriber<? super T> s);
}

可以看到，Publisher 根据收到的请求向当前订阅者 Subscriber 发送元素。

Subscriber

对应的，Subscriber 代表的是一种可以从发布者那里订阅并接收元素的订阅者。Subscriber 接口定义如下所示。

1
2
3
4
5
6

public interface Subscriber<T> {
    public void onSubscribe(Subscription s);
    public void onNext(T t);
    public void onError(Throwable t);
    public void onComplete();
}

我们注意到 Subscriber 接口包含了一组有用的方法，这组方法构成了数据流请求和处理的基本流程，我们一一来看一下。

其中，onSubscribe() 从命名上看就是一个回调方法，当发布者的 subscribe() 方法被调用时就会触发这个回调。而在该方法中有一个参数 Subscription，可以把这个 Subscription 看作是一种用于订阅的上下文对象。Subscription 对象中包含了这次回调中订阅者想要向发布者请求的数据个数。

当订阅关系已经建立，那么发布者就可以调用订阅者的 onNext() 方法向订阅者发送一个数据。这个过程是持续不断的，直到所发送的数据已经达到 Subscription 对象中所请求的数据个数。这时候 onComplete() 方法就会被触发，代表这个数据流已经全部发送结束。而一旦在这个过程中出现了异常，那么就会触发 onError() 方法，我们可以通过这个方法捕获到具体的异常信息进行处理，而数据流也就自动终止了。

Subscription

Subscription 代表的就是一种订阅上下文对象，它在订阅者和发布者之间进行传输，从而在两者之间形成一种契约关系。Subscription 接口定义如下所示。

1
2
3
4

public interface Subscription {
    public void request(long n);
    public void cancel();
}

这里的 request() 方法用于请求 n 个元素，订阅者可以通过不断调用该方法来向发布者请求数据；而 cancel() 方法显然是用来取消这次订阅。请注意，Subscription 对象是确保生产者和消费者针对数据处理速度达成一种动态平衡的基础，也是流量控制中实现背压机制的关键所在，我们可以通过下图来进一步理解整个数据请求和处理过程。

图 6 Subscription 与背压机制示意图

Publisher、Subscriber 和 Subscription 接口是响应式编程的核心组件，响应式流规范也只包含了这些接口，因此是一个非常抽象且精简的接口规范。结合前面的讨论结果，我们可以明确，响应式流规范实际上提供了一种“推-拉”结合的混合数据流模型。

当然，响应式流规范非常灵活，还可以提供独立的“推”模型和“拉”模型。如果为了实现纯“推”模型，我们可以考虑一次请求足够多的元素；而对于纯“拉”模型，相当于就是在每次调用 Subscriber 的 onNext() 方法时只请求一个新元素。

响应式流的技术生态圈

响应式流是一种规范，而该规范的核心价值就在于为业界提供了一种非阻塞式背压的异步流处理标准。各个供应商都可以基于该规范实现自己的响应式开发库，而这些开发库之间则可以做到互相兼容、相互交互。

目前，业界主流响应式开发库包括 RxJava、Akka、Vert.x 以及 Project Reactor。在本课程中，我们将重点介绍 Project Reactor，它是 Spring 5 中所默认集成的响应式开发库。

小结与预告

承接上一讲内容，本讲进一步分析了数据流的概念的分类，以及“推”流模式下的流量控制问题，从而引出了响应式系统中的背压机制。而流量控制的解决方案都包含在响应式流规范中，我们对这一规范中的核心组件展开了详细的说明。

响应式流规范是对响应式编程思想精髓的呈现，对于开发人员而言，理解这一规范有助于更好地掌握开发库的使用方法和基本原理。

这里给你留一道思考题：你能简要描述响应式流规范中数据的生产者和消费者之间的交互关系吗？

下一讲，我们来聊聊响应式编程的应用场景，相信这也是你最关心的内容，到时见。

点击链接，获取课程相关代码↓↓↓https://github.com/lagoueduCol/ReactiveProgramming-jianxiang.git

-– ### 精选评论 ##### *晓： > 打卡 ##### **海： > 啊，这。就是rxjava里面的知识点，这这里没想到和后端融合了 ##### **亮： > 订阅者通过subscribe方法订阅，发布者回调onSubscribe方法，通过Subscription沟通协调。发布订阅关系建立好后，发布者调用onNext方法，发送数据，直到完成发送的个数。发送完成会触发onComplate方法，异常则触发onError方法，数据流也会终止。 ######     讲师回复： >     对的 ##### *星： > 声音有点奇怪了，这是感冒了吗 ######     讲师回复： >     感谢关心，没事哈 ##### **9318： > 请问老师，订阅后，生产者把消息推给消费者，是有变化持续推送，还是推一次？ ######     讲师回复： >     是根据Subscription中的请求数量来确定的 ##### *瑾： > 老师，我现在的项目在每个处理复杂业务逻辑或者有外部api调用的节点出，都会进行publishon切换另一个单独线程，那这样每个节点于节点之间都需要做背压吗 ######     讲师回复： >     是的 ##### **乐： > 那意思就是消费者再调用request之前，生产者是一只阻塞得吗？假如有两个消费者，第一个消费者请求了1000条数据，第二个消费者一直没有请求。那当第二个消费者请求得时候还能看到那1000数据吗？如果能得话，是缓存在生产者哪里？会不会把生产者撑爆？ ######     讲师回复： >     不大明白你的意思，本质上请求-响应过程从原始的拉模式变成了推-拉模式，其他不变 ##### *刚： > 平台网关做的流控，跟老师讲的生产者、消费者感觉是相反的，流控是用来限制消费者的并发，避免太多的并发导致网关的异常。 ######     讲师回复： >     这是两个概念，背压是根据消费者的请求来控制生产者的速度 ##### **栋： > 老师，我有个问题。响应式关于流的处理，如果直接用mq来做，比如kafka，可不可以呢 ######     讲师回复： >     我认为是可以的，消息驱动本身实际上就可以认为是一种流处理机制