ProjectReactor响应式编程的探索与实践

总篇175篇 2022年第49篇1. 前言?在前后端分离和服务化的环境下，通常需要一个API网关统一对接C端、B端和运营等前端项目。这个API网关的主要作用有两个：一是为所有前端项目提供统一的入口，然后按照规则转发到后端服务上，之所以多一层转发，是因为服务化后的项目与前端是多对多的关系，一个前端项目可能会访问多个后端的项目，一个后端项目也可能对多个前端项目提供服务。二是对所有的前端项目进行统一鉴权，鉴权后将用户ID写入请求头并转发给后端项目，所有的后端项目都是无会话状态的，只需要根据ID对外提供接口。Spring Cloud Gateway 和 Netflix Zuul 都是这样一个API网关项目，能够按照业务规则实现请求的转发功能，也可以将统一鉴权很好的集成进去。最开始使用的 Zuul，和其他业务项目一样基于 Web Servlet 技术栈。随着 Spring 不断的升级，后来的项目也就从 Zuul 改成了 Spring Cloud Gateway。由于 Spring Cloud Gateway 构建在 Web Reactive技术栈之上，因此相关的业务就开始使用 WebFlux、WebClient 和 Reactor。2. Reactor 是什么?Reactor 是一个类库，是响应式编程范式的一个实现，主要用来开发非阻塞、异步应用，是 Spring 响应式生态系统的基础，同样的类库还有 RxJava。响应式编程范式定义了一组接口，作为观察者设计模式的扩展。核心接口如下:interface?Publisher?{void?subscribe(Subscriber?s);}interface?Subscriber?{void?onSubscribe(Subscription?s);void?onNext(T?t);void?onError(Throwable?t);void?onComplete();}interface?Subscription?{void?request(long?n);void?cancel();}Reactor 实现了这组接口，同时还提供了丰富的操作符，用于快速构建? Publisher 。在业务开发中，主要使用 Reactor 操作符，按照数据流的形式组合起来实现业务逻辑，就像下面这样:wechatWebClient.getUserInfo(suiteId,code).filter(it->!StringUtils.isEmpty(it.getCorpId())).switchIfEmpty(createUnauthorizedError("xxx")).doOnNext(token::setUserInfoResp).flatMap(it->wechatWebClient.getCorpInfo(suiteId,it.getCorpId())).filter(CorpInfoResp::isAuthenticated).switchIfEmpty(createUnauthorizedError("xxx")).doOnNext(it->token.setAgentId(it.getAuthorization().getAgentId())).flatMap(it->accountWebClient.getAccountByCorpId(it.getCorpId())).filter(it->!it.expired()).switchIfEmpty(createUnauthorizedError("xxx")).doOnNext(it->token.setCompanyId(String.valueOf(it.getId()))).thenReturn(RespBody.ok());3. 为什么要使用 Reactor?使用 Reactor 的主要原因是能够提高资源的利用效率，减少硬件资源的占用。响应式编程能用更少的资源做更多的工作，用更少的实例满足更多的并发连接。传统的开发方式使用阻塞代码编写程序，在出现性能问题之前，这种做法很好，一旦程序涉及大量的I/O操作(例如网络调用和数据库请求)，就会浪费系统资源，因为很多线程在等待数据时处于空闲状态。并且并发越大，问题就越明显。通过编写异步、非阻塞代码，可以有效的解决这些问题，然而JDK提供的异步API比较难用，因此也就催生了像 Reactor 这种响应式库。Reactor 这类响应式库与JDK的异步API相比，大大降低了异步编程的难度，同时还关注了代码的可组合性和可读性，并且可以使用丰富的运算符将数据作为流进行操作。4. Reactor 实践?下面示例使用 Groovy 语法描述：4.1创建元素简单的理解 Reactor 实际就两个核心类 Mono 和? Flux，他们都实现了 Publisher 接口，Mono 包含0-1个元素，Flux 包含0-N个元素，List 不一定要使用 Flux。def?mono?=?Mono.just(["foo",?"bar",?"foobar"]).map({?it.collect?{?item?->?item.toUpperCase()?}?})mono.subscribe({?println("the?mono?data?is?$it")?})def?flux?=?Flux.fromIterable(["foo",?"bar",?"foobar"]).map({?it.toUpperCase()?})flux.subscribe({?println("the?flux?data?is?$it")?})the mono data is [FOO, BAR, FOOBAR]the flux data is FOOthe flux data is BARthe flux data is FOOBARjust 传递的是对象，在声明时会立即初始化，?defer?传递的是函数，只有在订阅后才执行，无论何时都应该优先使用 defer。def?data?=?"foo"def?pub1?=?Mono.just(data)data?=?"bar"pub1.subscribe({?println("just?data?is?$it")?})def?pub2?=?Mono.defer({?Mono.just(data)?})data?=?"foobar"pub2.subscribe({?println("defer?data?is?$it")?})just data is foodefer data is foobar4.2?转换元素map 同步转换元素， flatMap 异步转换元素，前者直接在内存中操作，后者大多情况会有I/O。Mono.just("foobar").map({?it.size()?}).subscribe({?println("map?result?is?$it")?})Mono.just("foobar").flatMap({?Mono.just(it.size())?}).subscribe({?println("flatMap?result?is?$it")?})map result is 6flatMap result is 64.3附加行为按照响应式编程范式，在订阅后正常会相继执行以下操作:- Subscriber.onSubscribe()- Subscription.request()- Subscriber.onNext()- Subscriber.onComplete()doOn* 这组附加行为就是在执行到这些阶段时被调用，例如? doOnNext 是在执行完? Subscriber.onNext() 后触发附加行为不改变序列的元素，是在订阅执行到某个阶段时，触发一些自定义的行为：Flux.fromIterable([3,?4,?5]).doOnNext({?println("init?value?is?$it")?}).map({?it?*?3?}).doOnNext({?println("map?value?is?$it")?}).doOnComplete({?println("flux?completed")?}).subscribe()init value is 3map value is 9init value is 4map value is 12init value is 5map value is 15flux completed?log 可以说是一个特殊的附加行为，可以输出订阅后每个阶段的执行过程：Flux.fromIterable([3,?4,?5]).log().map({?it?*?3?}).log().subscribe()reactor.Flux.Iterable.1 - | onSubscribe([Synchronous Fuseable] FluxIterable.IterableSubscription)reactor.Flux.MapFuseable.2 - | onSubscribe([Fuseable] FluxMapFuseable.MapFuseableSubscriber)reactor.Flux.MapFuseable.2 - | request(unbounded)reactor.Flux.Iterable.1 - | request(unbounded)reactor.Flux.Iterable.1 - | onNext(3)reactor.Flux.MapFuseable.2 - | onNext(9)reactor.Flux.Iterable.1 - | onNext(4)reactor.Flux.MapFuseable.2 - | onNext(12)reactor.Flux.Iterable.1 - | onNext(5)reactor.Flux.MapFuseable.2 - | onNext(15)reactor.Flux.Iterable.1 - | onComplete()reactor.Flux.MapFuseable.2 - | onComplete()4.4切换序列?then 结束上一个序列后，切换到下一个序列，两个序列没有必要的相关性时使用。比如执行完某项任务后返回一个成功。Mono.just("foo").doOnNext({?println("do?something?$it")?}).then(Mono.just("success")).subscribe({?println("then?result?is?$it")?})do something foothen result is successthenReturn 是 then 的简化操作。Mono.just("bar").doOnNext({?println("do?something?$it")?}).thenReturn("success").subscribe({?println("then?return?is?$it")?})do something barthen return is successswitchIfEmpty 当前序列没有元素时，切换到另一个序列。Flux.fromIterable([3,?4,?5]).filter({?it?>?5?}).switchIfEmpty(Flux.just(6)).map({?it?+?1?}).subscribe({?println("switchIfEmpty?result?is?$it")?})switchIfEmpty result is 74.5错误处理error 抛出异常 onErrorResume 捕获异常做后续处理。Flux.fromIterable([3,?4,?5]).filter({?it?>?5?}).switchIfEmpty(Mono.error(new?RuntimeException("value?too?small"))).onErrorResume({println("on?error?resume?${it.getMessage()}")Flux.fromIterable([6,?7,?8])}).subscribe({?println("flux?value?is?$it")?})on error resume value too smallflux value is 6flux value is 7flux value is 8onErrorReturn捕获异常并返回值, 是? onErrorResume 的简化操作。Mono.defer({?Mono.just(null)?}).onErrorReturn(NullPointerException,?"foobar").subscribe({?println("on?error?return?$it")?})on error return foobaronErrorMap 捕获异常并重新抛出另一个异常。Mono.defer({?Mono.just(null)?}).onErrorMap(NullPointerException,?{new?RuntimeException("null?pointer")}).doOnError({println("error?message?${it.getMessage()}")}).subscribe()error message null pointer4.6传递上下文在响应式编程中, 一个线程可以大致同时处理多个异步序列, 一个异步序列也可以从一个线程切换到另一个线程。像 ThreadLocal 这种允许将数据与线程相关联的方式，在响应式编程中是不适用的， Reactor 提供了一个高级特性 Context 来解决这个问题：Mono.just("foo").flatMap({Mono.subscriberContext().map({?ctx?->?it?+?ctx.get("key")?})}).subscriberContext({?Context.of("key",?"bar")?}).subscribe({?println("context?result?is?$it")?})context result is foobarsubscriberContext 的链是从下往上流动的, 所以需要先读取后写入。Mono.just("foo").subscriberContext({?Context.of("key",?"bar")?}).flatMap({Mono.subscriberContext().map({?ctx?->?it?+?ctx.getOrDefault("key",?"")?})}).subscribe({?println("context?result?is?$it")?})context result is foo多次向 Context 中写入相同 Key 时, 按照顺序读取其下最接近的。Mono.just("hello").flatMap({Mono.subscriberContext().map({?ctx?->?it?+?"?"?+?ctx.get("key")?})}).subscriberContext({?it.put("key",?"foo")?}).flatMap({Mono.subscriberContext().map({?ctx?->?it?+?ctx.get("key")?})}).subscriberContext({?it.put("key",?"bar")?}).subscribe({?println("context?result?is?$it")?})context result is hello foobar4.7单元测试Reactor 无法进行常规的断言，因此提供了专门的测试类库，使用 StepVerifier 测试场景。when:def?mono?=?Mono.just("foo").map({?it?+?"bar"?})then:StepVerifier.create(mono).expectNext("foobar").verifyComplete()下面使用 expectNext 断言元素，使用 expectError 断言异常，只有在调用 verify... 之后才执行验证，不调用不执行。when:def?flux?=?Flux.just("thing1",?"thing2").concatWith(Mono.error(new?RuntimeException()))then:StepVerifier.create(flux).expectNext("thing1").expectNext("thing2").expectError().verify()5. 使用 Reactor 的注意事项?5.1代码为什么没有执行下面 { it + "bar" } 不会被执行，操作符的链式调用每次都会返回一个新创建的对象。def?mono?=?Mono.just("foo")mono.map({?it?+?"bar"?})mono.subscribe({?println("the?result?is?$it")?})the result is foo下面 println() 不会被执行，Mono 不订阅不执行，最外层的 subscribe 不会传播到 doOnNext 的 Mono 里面，同理 doOn* 都不应该执行 Publisher 相关的代码。Mono.just("foo1").doOnNext({?item?->Mono.fromSupplier({?println("do?something?$item")?})}).subscribe()下面 println() 可以执行，但是对外层的订阅者来说，就丧失了对内的控制权，因此不建议有内部订阅。Mono.just("bar").doOnNext({?item?->Mono.fromSupplier({?println("do?something?$item")?}).subscribe()}).subscribe()do something bar为了将订阅在内部传播下去，正确的方法应该使用? flatMap()Mono.just("foo2").flatMap({?item?->Mono.fromSupplier({?println("do?something?$item")?})}).subscribe()do something foo25.2NULL问题?下面 onErrorReturn 是无效的，Mono.just()?不接收空值，在构建阶段就会抛出异常，而捕获异常是在订阅阶段发生的。when:Mono.just(null).onErrorReturn("foo").subscribe({?println("the?result?is?$it")?})then:thrown(NullPointerException)?下面 onErrorReturn 是有效的，这就是上文中提到的为什么要优先使用 defer 的原因，同样的异常会在订阅执行后被捕获。when:Mono.defer({ Mono.just(null) }) .onErrorReturn("bar") .subscribe({ println("the result is $it") })then:noExceptionThrown()the result is bar如果没有明确的表示，所有的调用都不能传递? NULL，即便是可以传递空的 justOrEmpty 也是判断空后返回 Mono.empty()，下面的 justOrEmpty 和 empty 是等价的。Mono.justOrEmpty(null).log().subscribe()Mono.empty().log().subscribe()reactor.Mono.Empty.1 - onSubscribe([Fuseable] Operators.EmptySubscription)reactor.Mono.Empty.1 - request(unbounded)reactor.Mono.Empty.1 - onComplete()reactor.Mono.Empty.2 - onSubscribe([Fuseable] Operators.EmptySubscription)reactor.Mono.Empty.2 - request(unbounded)reactor.Mono.Empty.2 - onComplete()因此 Reactor 的设计原则是不要在序列中传递空值，任何 NULL 参数都会抛出 NPE。Mono.just("foo").map({?null?}).subscribe()Caused by: java.lang.NullPointerException: The mapper returned a null value. at java.util.Objects.requireNonNull(Objects.java:228) at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107) at reactor.core.publisher.Operators$ScalarSubscription.request(Operators.java:2393)5.3通过异常信息无法定位问题Reactor 默认没有启用调试功能，通过堆栈跟踪信息排查和定位问题比较困难。Caused by: java.lang.IndexOutOfBoundsException: Source emitted more than one item at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:127) at reactor.core.publisher.FluxFlatMap$FlatMapMain.tryEmitScalar(FluxFlatMap.java:481) at reactor.core.publisher.FluxFlatMap$FlatMapMain.onNext(FluxFlatMap.java:414) at reactor.core.publisher.FluxRange$RangeSubscription.slowPath(FluxRange.java:154) at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:109) at reactor.core.publisher.FluxFlatMap$FlatMapMain.onSubscribe(FluxFlatMap.java:364) at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68)下面是启用了调试功能的堆栈跟踪信息，可以明确是哪一行代码出现了问题。Caused by: java.lang.IndexOutOfBoundsException: Source emitted more than one item at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:127) Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: Assembly trace from producer [reactor.core.publisher.MonoSingle] : reactor.core.publisher.Flux.single(Flux.java:7887) sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)Error has been observed at the following site(s): |_ Flux.single ? at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) |_ Mono.map ? at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) |_ Mono.filter ? at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)Reactor 提供了三种用于增强调试信息的方法:1.全局启用调试功能 Hooks.onOperatorDebug() 这是最简单的方式也是最慢的；2.在调用链中使用 checkpoint() 手动激活堆栈跟踪；3.使用生产级别的全局调试类库 reactor-tools。6. 总结?在实践过程中，由于编程方式的不同，使用声明式编程不论是编码、调试和维护，相比命令式编程门槛要高一些。因此梳理了一些在使用过程中容易产生疑惑的问题，希望对其他初学者有所帮助。作者简介苏彬彬经销商技术部-商业资源团队2017年加入汽车之家经销商事业部，目前主要负责事业部内创新商业项目的研发工作，热衷于业内新技术的探索与实践。阅读更多：细粒度分类算法在汽车之家车系识别业务中的应用DDD在经销商的应用Flutter沉浸式复杂吸顶交互页面结构设计与实践