关于后端:Backpressure-in-Reactive-Systems-响应式系统的反压

6次阅读

共计 2379 个字符,预计需要花费 6 分钟才能阅读完成。

原文
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 0 到 N 号元素的流。反对 Reactive- 流和反压
Observable 0 到 N 元素的流。不反对反压
Single一个准确的流:1 个元素或一个谬误
Maybe一个包含以下的流:没有元素 一个元素 或一个谬误
Completable一个流没有元素但:是一个 completion 完结或一个谬误的信号

在这些类中,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

正文完
 0