Reactive系统的反压

大家好，欢迎来到IT知识分享网。

原文 Backpressure in Reactive Systems

https://foojay.io/today/backpressure-in-reactive-systems/?spm=ata..0.0.eDx9MZ

翻译 时序

图片来自：Pexels 上的 Dave Mungai 拍摄的图片

一月份中旬，我基于我的文章迁移到Reactive的必要条件Spring Boot应用[1]做了一个分享 https://www.youtube.com/watch?v=w0b4OmhBI

因为那是一个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框架都有提供方案，我们来看下。

RxJava3的反压

RxJava v3提供以下基础类：

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

反压并没有解决Observable过度生成或Subscriber过度消费。它只是将这个问题从处理的链条中移动到了一个比较好处理的地方。–响应式进行反压不是万金油。

为了解决这个，RxJava提供处理“过度生产“元素的两个主要策略：

•将元素存储到一个缓存里，如果没有足够的缓存，可能会产生OutOfMemoryError。

•丢掉数据

下图描述了这些策略的不同实现方法：

记住onBackPressureLatest操作同使用onBackpressureBuffer(1)类似：

这张图来自RxJava的Wiki。

与其他框架不同的是，RxJava提供方法来在发送完所有元素后发送溢出异常信号。这让消费者可以收到数据而同时清楚发送方已经丢了数据。

ProjectReactor中的反压

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-streams/reactive-streams-jvm

How (not) to use Reactive Streams in Java 9+

https://blog.softwaremill.com/how-not-to-use-reactive-streams-in-java-9-7a39ea9c2cb3

RxJava Backpressure

https://github.com/ReactiveX/RxJava/blob/3.x/docs/Backpressure.md

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

开发者/科幻爱好者/硬核主机玩家/业余翻译

转载请注明。

B站: https://space.bilibili.com//

交流Email: [2]

References

[1] 迁移到Reactive的必要条件Spring Boot应用: https://hazelcast.org/blog/migrating-from-imperative-to-reactive/
[2] : mailto: