原文
https://foojay.io/today/backp…

一月中，我基于我的文章迁徙到Reactive的必要条件Spring Boot利用做了一个分享
https://www.youtube.com/watch…

因为那是一个Kotlin的团聚，我是用Kotlin代码展现的，同时我加了一个将代码库迁徙到协程的步骤。

在QA环节，有人问到是否协程实现了反压。我抵赖我也不确定，所以我做了一点钻研。

本文提供了对于反压的概要信息，还有如何用Rxjava(v3)，Project Reactor和Kotlin的协程Coroutines如何解决。

什么是反压？

反压是指对管道中流体的抵挡或反向作用力，导致丢失摩擦力和压力升高。反压的说法不太失当，压力是个标量，有大小，但没有方向 — 维基百科
在软件中，反压跟这有点关系但也有不同的含意：假如有一个很快的数据发送方和一个比较慢的数据接管方，反压是指一种机制能够反向推动发送方不要把接管方压垮。

无论是reactivestreams.org或java.until.concurrent.Flow，反馈流都提供以下四个构建块

• Publisher发送元素
• Subscriber对收到的元素产生反馈
• 一个Subscription来绑定Publisher和Subscriber
• 一个Processor
  这是类图：

Subscription的request()办法是反压的顶层。
标准很直白：

Subscriber必须通过Subscription.request(long n)来发送需要信号后接管onNext信号。这里隐含的规定就是由Subscriber决定什么时候和有多少元素须要被接管。为了防止可重入Subscription办法引起的信号重排序，强烈推荐Subscriber办法的实现在调用Subscription办法的最初对任何信号处理都是用同步的形式。举荐Subscriber申请它们能够解决的下限，因为一次只申请一个元素会导致低效的“进行和期待”协定。 — JVM的Reactive流标准

响应流的标准很规范。它们也有基于Java的TCK。

但要定义如何治理producer发送上游无奈解决的元素就超出这个标准的范畴了。问题比较简单，解决办法也多。每种Reactive框架都有提供计划，咱们来看下。

RxJava 3的反压

RxJava v3提供以下根底类：

在这些类中，Flowable是惟一实现了Reactive流-反压的流。因而，提供反压不是惟一的问题。RxJava wiki指出：

反压并没有解决Observable适度生成或Subscriber适度生产。它只是将这个问题从解决的链条中挪动到了一个比拟好解决的中央。 –响应式进行反压不是万金油。

为了解决这个，RxJava提供解决“适度生产“元素的两个次要策略：

• 将元素存储到一个缓存里，如果没有足够的缓存，可能会产生OutOfMemoryError。
• 丢掉数据
  下图形容了这些策略的不同实现办法：

记住onBackPressureLatest操作同应用onBackpressureBuffer(1)相似：

这张图来自RxJava的Wiki。

与其余框架不同的是，RxJava提供办法来在发送完所有元素后发送溢出异样信号。这让消费者能够收到数据而同时分明发送方曾经丢了数据。

Project Reactor中的反压

Project Reactor中提供的策略与RxJava相似。

API有点不一样。比方，如果生产者溢出Project Reactor提供一个不便的办法来抛异样：

var stream = Stream.generate(Math::random);

// RxJava
Flowable.fromStream(stream) // 1
.onBackpressureBuffer(0); // 2

// Project Reactor
Flux.fromStream(stream) // 1
.onBackpressureError(); // 2

• 创立Reactive流
• 如果生产者溢出抛异样

上面是高亮了反压能力的Flux类图：

与其余框架相比，Project Reactor提供设置缓存TTL的办法来避免溢出。

协程中的反压

协程提供同样的缓存和生效能力。协程的根底类是Flow。

你能够这样应用：

flow { // 1
while (true) emit(Math.random()) // 2
}.buffer(10)

• 建一个Flow类，由上面定义content
• 定义Flow的内容
• 设置缓存容量为10

论断

RxJava，Project Reactor，Kotlin协程都提供反压能力。在生产者比消费者更快时提供两种策略：缓存数据或摈弃数据。

更多：

Reactive Streams JVM specifications
https://github.com/reactive-s…
How (not) to use Reactive Streams in Java 9+
https://blog.softwaremill.com…
RxJava Backpressure
https://github.com/ReactiveX/…

本文来自祝坤荣(时序)的微信公众号「麦芽面包」，公众号id「darkjune\_think」

开发者/科幻爱好者/硬核主机玩家/业余翻译
转载请注明。

B站: https://space.bilibili.com/23…
交换Email: zhukunrong@yeah.net