RxJava 变换操作符完全指南：从 map、flatMap 到 buffer、window 的源码级解析

RxJava 变换操作符完全指南从 map、flatMap 到 buffer、window 的源码级解析【免费下载链接】RxJavaRxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.项目地址: https://gitcode.com/gh_mirrors/rx/RxJava本指南以仓库内 docs/Transforming-Observables.md 为骨架系统梳理 RxJava 中全部变换类操作符。你将掌握map/cast的一对一投影、flatMap家族的并发合并、concatMap家族的串行拼接、switchMap的只取最新、scan的累积扫描、groupBy的分组、buffer/window的批量聚合以及每个操作符在Flowable/Observable/Maybe/Single/Completable上的可用范围并理解它们在 Observable.java、Flowable.java 中的底层实现与对应测试。什么是变换Transforming操作符变换操作符用于对响应式源如Observable、Flowable发射的每一个数据项进行加工、展开、聚合或重组再以新的形态发射出去。它们是构建异步数据管道时最常用的操作符族——无论是把用户 ID 映射成用户详情、把事件流按时间批量落库还是把多条数据流拼接成一条都离不开它们。在阅读示例之前需要注意一个仓库细节当前仓库的根包名是io.reactivex.rxjava4参见 Observable.java 的package io.reactivex.rxjava4.core;文档中出现的io.reactivex.functions.Function、io.reactivex.CompletableSource等类型对应到本仓库分别为io.reactivex.rxjava4.functions.Function与io.reactivex.rxjava4.core.CompletableSource。为忠实保留原文档语义下文示例沿用文档写法实际编译时按根包名替换即可。变换操作符一览与可用性速查以下表格汇总了全部变换操作符及其在五大反应式类型上的可用性✓ 表示可用✗ 表示不可用方便你在选型时快速定位操作符FlowableObservableMaybeSingleCompletablebuffer✓✓✗✗✗cast✓✓✓✓✗concatMap/concatMapDelayError/concatMapEager/concatMapEagerDelayError✓✓✗✗✗concatMapCompletable/concatMapCompletableDelayError✓✓✗✗✗concatMapIterable✓✓✗✗✗concatMapMaybe/concatMapMaybeDelayError✓✓✗✗✗concatMapSingle/concatMapSingleDelayError✓✓✗✗✗flatMap✓✓✓✓✗flatMapCompletable✓✓✗✗✗flatMapIterable✓✓✗✗✗flatMapMaybe✓✓✗✓✗flatMapObservable✗✗✓✓✗flatMapPublisher✗✗✓✓✗flatMapSingle✓✓✓✗✗flatMapSingleElement✗✗✓✗✗flattenAsFlowable/flattenAsObservable✗✗✓✓✗groupBy✓✓✗✗✗map✓✓✓✓✗scan✓✓✗✗✗switchMap✓✓✗✗✗window✓✓✗✗✗核心规律凡是要求“每个数据项都产出多个结果”的操作如map、cast、flatMap基本都支持Maybe/Single凡是涉及多数据项之间聚合、分组、串并关系的操作如buffer、window、groupBy、scan只存在于多值流Flowable/Observable上Completable因不发射数据项因此不提供任何变换操作符。一对一映射map 与 castmap逐项应用函数map对源发射的每个数据项应用给定的io.reactivex.functions.Function并发射函数计算结果。它是变换家族中最基础、最常用的操作符在 Observable.java 附近的flatMap以及各操作符实现中随处可见其身影。Observable.just(1, 2, 3) .map(x - x * x) .subscribe(System.out::println); // prints: // 1 // 4 // 9cast按类型转换cast把源发射的每个数据项强制转换为指定类型后发射。它本质上等价于map内部的类型转换但在类型不匹配时会抛ClassCastException。在 Observable.java 中cast(ClassU clazz)被定义为一个泛型方法返回ObservableUObservable还有一个便捷的ofType组合见 Observable.java它先按类型过滤再cast。ObservableNumber numbers Observable.just(1, 4.0, 3f, 7, 12, 4.6, 5); numbers.filter((Number x) - Integer.class.isInstance(x)) .cast(Integer.class) .subscribe((Integer x) - System.out.println(x)); // prints: // 1 // 7 // 12 // 5对应测试可见 ObservableCastTest.java 与 ObservableMapTest.java。并发合并flatMap 家族flatMap的核心语义是对源发射的每个数据项应用一个返回反应式源的函数然后把所有函数产生的源合并merge发射。合并是并发的——各个内部源的发射会相互交错因此输出顺序不确定。flatMap合并任意反应式源Observable.just(A, B, C) .flatMap(a - { return Observable.intervalRange(1, 3, 0, 1, TimeUnit.SECONDS) .map(b - ( a , b )); }) .blockingSubscribe(System.out::println); // prints (not necessarily in this order): // (A, 1) // (C, 1) // (B, 1) // (A, 2) // (C, 2) // (B, 2) // (A, 3) // (C, 3) // (B, 3)注意打印结果中(A, 1)、(C, 1)、(B, 1)的顺序是任意的这正是“合并”与“拼接”的本质区别。从源码看Observable.java 中的flatMap(mapper)默认使用StandardConcurrentBufferedConfig.MAX_DEFAULT配置即无最大并发限制的并发缓冲配置实际组装的是internal/operators/observable包下的合并类。flatMapCompletable只关心完成不关心数据flatMapCompletable要求映射函数返回io.reactivex.CompletableSource并返回一个在所有源都完成后才完成的Completable。它适合“对每个数据项执行副作用操作如写库、发消息但不需要结果值”的场景。ObservableInteger source Observable.just(2, 1, 3); Completable completable source.flatMapCompletable(x - { return Completable.timer(x, TimeUnit.SECONDS) .doOnComplete(() - System.out.println(Info: Processing of item \ x \ completed)); }); completable.doOnComplete(() - System.out.println(Info: Processing of all items completed)) .blockingAwait(); // prints: // Info: Processing of item 1 completed // Info: Processing of item 2 completed // Info: Processing of item 3 completed // Info: Processing of all items completedflatMapIterable展开同步集合当映射函数返回的是同步的java.lang.Iterable而非反应式源时用flatMapIterable更轻量——无需为每个元素创建内部源。Observable.just(1, 2, 3, 4) .flatMapIterable(x - { switch (x % 4) { case 1: return List.of(A); case 2: return List.of(B, B); case 3: return List.of(C, C, C); default: return List.of(); } }) .subscribe(System.out::println); // prints: // A // B // B // C // C // CflatMapMaybe与 Maybe 合并flatMapMaybe要求映射函数返回io.reactivex.MaybeSource并合并发射这些MaybeSource的结果。Maybe可以“为空”empty为空的Maybe不会贡献任何数据项。Observable.just(9.0, 16.0, -4.0) .flatMapMaybe(x - { if (x.compareTo(0.0) 0) return Maybe.empty(); else return Maybe.just(Math.sqrt(x)); }) .subscribe( System.out::println, Throwable::printStackTrace, () - System.out.println(onComplete)); // prints: // 3.0 // 4.0 // onCompleteflatMapSingle与 Single 合并Observable.just(4, 2, 1, 3) .flatMapSingle(x - Single.timer(x, TimeUnit.SECONDS).map(i - x)) .blockingSubscribe(System.out::print); // prints 1234上面示例中虽然每个Single的延迟时间不同4、2、1、3 秒但输出顺序并非严格的完成顺序而是发射交错后的结果因此得到1234。注意Maybe::flatMapSingle的语义差异当Maybe源为空时返回的Single会发出错误通知而不是静默完成MaybeObject emptySource Maybe.empty(); SingleObject result emptySource.flatMapSingle(x - Single.just(x)); result.subscribe( x - System.out.println(onSuccess will not be printed!), error - System.out.println(onError: Source was empty!)); // prints: // onError: Source was empty!如果希望“源为空则直接完成”而不是报错应改用Maybe::flatMapSingleElement其返回Maybe。flatMapObservable 与 flatMapPublisher从 Maybe/Single 展开为多值流这两个操作符把Maybe或Single发射的单个数据项交给映射函数展开成一个ObservableSourceflatMapObservable或org.reactivestreams.PublisherflatMapPublisher分别返回Observable与Flowable。适合“一个值拆成多个值”的场景例如把 CSV 字符串拆成多行。SingleString source Single.just(Kirk, Spock, Chekov, Sulu); ObservableString names source.flatMapObservable(text - { return Observable.fromArray(text.split(,)) .map(String::strip); }); names.subscribe(name - System.out.println(onNext: name)); // prints: // onNext: Kirk // onNext: Spock // onNext: Chekov // onNext: SuluSingleString source Single.just(Kirk, Spock, Chekov, Sulu); FlowableString names source.flatMapPublisher(text - { return Flowable.fromArray(text.split(,)) .map(String::strip); }); names.subscribe(name - System.out.println(onNext: name)); // prints: // onNext: Kirk // onNext: Spock // onNext: Chekov // onNext: SuluflatMapSingleElementMaybe 版的不报错展开flatMapSingleElement仅存在于Maybe上映射函数返回io.reactivex.SingleSource若源Maybe有值则发射该Single的结果若源Maybe为空则直接完成绝不报错。MaybeInteger source Maybe.just(-42); MaybeInteger result source.flatMapSingleElement(x - { return Single.just(Math.abs(x)); }); result.subscribe(System.out::println); // prints 42flattenAsFlowable 与 flattenAsObservable把单个值展平为集合流与flatMapObservable/flatMapPublisher类似但映射函数返回同步的java.lang.Iterable分别产出Flowable与Observable。SingleDouble source Single.just(2.0); FlowableDouble flowable source.flattenAsFlowable(x - { return List.of(x, Math.pow(x, 2), Math.pow(x, 3)); }); flowable.subscribe(x - System.out.println(onNext: x)); // prints: // onNext: 2.0 // onNext: 4.0 // onNext: 8.0SingleDouble source Single.just(2.0); ObservableDouble observable source.flattenAsObservable(x - { return List.of(x, Math.pow(x, 2), Math.pow(x, 3)); }); observable.subscribe(x - System.out.println(onNext: x)); // prints: // onNext: 2.0 // onNext: 4.0 // onNext: 8.0串行拼接concatMap 家族concatMap与flatMap的唯一区别是拼接concat而非合并merge内部源按顺序一个接一个地订阅前一个源完成后才开始下一个因此输出顺序与输入顺序严格一致。代价是吞吐量低于flatMap。concatMap按顺序拼接Observable.range(0, 5) .concatMap(i - { long delay Math.round(Math.random() * 2); return Observable.timer(delay, TimeUnit.SECONDS).map(n - i); }) .blockingSubscribe(System.out::print); // prints 01234示例中每个内部Observable.timer的延迟是随机的但输出依然是严格的01234这正是串行拼接的证明。concatMapIterable拼接同步集合Observable.just(A, B, C) .concatMapIterable(item - List.of(item, item, item)) .subscribe(System.out::print); // prints AAABBBCCCconcatMapCompletable 与 concatMapCompletableDelayErrorconcatMapCompletable要求映射函数返回io.reactivex.CompletableSource逐个订阅全部完成后返回的Completable才完成ObservableInteger source Observable.just(2, 1, 3); Completable completable source.concatMapCompletable(x - { return Completable.timer(x, TimeUnit.SECONDS) .doOnComplete(() - System.out.println(Info: Processing of item \ x \ completed)); }); completable.doOnComplete(() - System.out.println(Info: Processing of all items completed)) .blockingAwait(); // prints: // Info: Processing of item 2 completed // Info: Processing of item 1 completed // Info: Processing of item 3 completed // Info: Processing of all items completed注意这里完成顺序是2、1、3——与flatMapCompletable示例中1、2、3的并发完成顺序形成鲜明对比concatMap严格按输入顺序逐个处理即便前一项耗时更长。concatMapCompletableDelayError与之相同但延迟错误某个源出错不会中断后续处理所有源都终止后错误才被统一上报ObservableInteger source Observable.just(2, 1, 3); Completable completable source.concatMapCompletableDelayError(x - { if (x.equals(2)) { return Completable.error(new IOException(Processing of item \ x \ failed!)); } else { return Completable.timer(1, TimeUnit.SECONDS) .doOnComplete(() - System.out.println(Info: Processing of item \ x \ completed)); } }); completable.doOnError(error - System.out.println(Error: error.getMessage())) .onErrorComplete() .blockingAwait(); // prints: // Info: Processing of item 1 completed // Info: Processing of item 3 completed // Error: Processing of item 2 failed!可以看到尽管数据项2的处理立即失败但数据项1、3仍被正常处理完毕错误最后才上报。concatMapDelayError延迟错误的串行拼接concatMapDelayError在保持串行语义的同时延迟所有内部源的错误直到所有源终止Observable.intervalRange(1, 3, 0, 1, TimeUnit.SECONDS) .concatMapDelayError(x - { if (x.equals(1L)) return Observable.error(new IOException(Something went wrong!)); else return Observable.just(x, x * x); }) .blockingSubscribe( x - System.out.println(onNext: x), error - System.out.println(onError: error.getMessage())); // prints: // onNext: 2 // onNext: 4 // onNext: 3 // onNext: 9 // onError: Something went wrong!concatMapEager急切订阅的串行拼接concatMapEager与concatMap输出顺序一致但急切地eagerly订阅所有内部源——各内部源并行开始执行因此doOnNext的完成日志乱序只是发射到下游时仍按顺序拼接Observable.range(0, 5) .concatMapEager(i - { long delay Math.round(Math.random() * 3); return Observable.timer(delay, TimeUnit.SECONDS) .map(n - i) .doOnNext(x - System.out.println(Info: Finished processing item x)); }) .blockingSubscribe(i - System.out.println(onNext: i)); // prints (lines beginning with Info... can be displayed in a different order): // Info: Finished processing item 2 // Info: Finished processing item 0 // onNext: 0 // Info: Finished processing item 1 // onNext: 1 // onNext: 2 // Info: Finished processing item 3 // Info: Finished processing item 4 // onNext: 3 // onNext: 4观察输出Info...日志乱序2 比 0 先完成但onNext严格按0, 1, 2, 3, 4顺序输出。concatMapEager的语义是“执行并行、输出串行”适合内部源耗时较长、又想保持输出顺序的场景。在仓库中Observable与Flowable均有对应实现见 Observable.java 中的concatMapEager系列与 ObservableConcatMapEagerTest.java。concatMapEagerDelayError可配置错误时序的急切拼接concatMapEagerDelayError额外接收一个boolean参数为true时所有源的错误都延迟到末尾统一上报为false时主源的错误会在当前内部源终止后立即上报。ObservableInteger source Observable.create(emitter - { emitter.onNext(1); emitter.onNext(2); emitter.onError(new Error(Fatal error!)); }); source.doOnError(error - System.out.println(Info: Error from main source error.getMessage())) .concatMapEagerDelayError(x - { return Observable.timer(1, TimeUnit.SECONDS).map(n - x) .doOnSubscribe(it - System.out.println(Info: Processing of item \ x \ started)); }, true) .blockingSubscribe( x - System.out.println(onNext: x), error - System.out.println(onError: error.getMessage())); // prints: // Info: Processing of item 1 started // Info: Processing of item 2 started // Info: Error from main source Fatal error! // onNext: 1 // onNext: 2 // onError: Fatal error!concatMapMaybe 与 concatMapMaybeDelayErrorconcatMapMaybe要求映射函数返回io.reactivex.MaybeSource按顺序拼接这些MaybeSource的发射结果Observable.just(5, 3,14, 2.71, FF) .concatMapMaybe(v - { return Maybe.fromCallable(() - Double.parseDouble(v)) .doOnError(e - System.out.println(Info: The value \ v \ could not be parsed.)) // Ignore values that can not be parsed. .onErrorComplete(); }) .subscribe(x - System.out.println(onNext: x)); // prints: // onNext: 5.0 // Info: The value 3,14 could not be parsed. // onNext: 2.71 // Info: The value FF could not be parsed.concatMapMaybeDelayError在保持串行拼接的同时延迟错误DateTimeFormatter dateFormatter DateTimeFormatter.ofPattern(dd.MM.uuuu); Observable.just(04.03.2018, 12-08-2018, 06.10.2018, 01.12.2018) .concatMapMaybeDelayError(date - { return Maybe.fromCallable(() - LocalDate.parse(date, dateFormatter)); }) .subscribe( localDate - System.out.println(onNext: localDate), error - System.out.println(onError: error.getMessage())); // prints: // onNext: 2018-03-04 // onNext: 2018-10-06 // onNext: 2018-12-01 // onError: Text 12-08-2018 could not be parsed at index 2注意格式非法的12-08-2018虽然中途报错但后续的06.10.2018、01.12.2018仍然被正常解析输出错误被延迟到最后统一上报。concatMapSingle 与 concatMapSingleDelayErrorconcatMapSingle要求映射函数返回io.reactivex.SingleSource按顺序拼接Observable.just(5, 3,14, 2.71, FF) .concatMapSingle(v - { return Single.fromCallable(() - Double.parseDouble(v)) .doOnError(e - System.out.println(Info: The value \ v \ could not be parsed.)) // Return a default value if the given value can not be parsed. .onErrorReturnItem(42.0); }) .subscribe(x - System.out.println(onNext: x)); // prints: // onNext: 5.0 // Info: The value 3,14 could not be parsed. // onNext: 42.0 // onNext: 2.71 // Info: The value FF could not be parsed. // onNext: 42.0concatMapSingleDelayError则延迟所有源错误DateTimeFormatter dateFormatter DateTimeFormatter.ofPattern(dd.MM.uuuu); Observable.just(24.03.2018, 12-08-2018, 06.10.2018, 01.12.2018) .concatMapSingleDelayError(date - { return Single.fromCallable(() - LocalDate.parse(date, dateFormatter)); }) .subscribe( localDate - System.out.println(onNext: localDate), error - System.out.println(onError: error.getMessage())); // prints: // onNext: 2018-03-24 // onNext: 2018-10-06 // onNext: 2018-12-01 // onError: Text 12-08-2018 could not be parsed at index 2只取最新switchMapswitchMap对源发射的每个数据项应用函数生成内部源但只发射最近一个内部源的结果每当新数据项到达前一个内部源会被退订取消订阅。非常适合搜索框输入联想、竞态消除等“以最新输入为准”的场景。Observable.interval(0, 1, TimeUnit.SECONDS) .switchMap(x - { return Observable.interval(0, 750, TimeUnit.MILLISECONDS) .map(y - x); }) .takeWhile(x - x 3) .blockingSubscribe(System.out::print); // prints 001122外层源每 1 秒发射一个新值内层源每 750ms 重复发射当前值。由于内层源每 750ms 才发射一次、而外层每秒切换一次输出00后切换到1、11后切换到2最终得到001122。switchMap在 Observable.java 中默认使用StandardBufferedConfig.DEFAULT底层由internal/operators/observable的 switch 类实现对应测试 ObservableSwitchTest.java。累积扫描scanscan使用io.reactivex.functions.BiFunction从种子值seed开始把“上一步的结果”与“下一个数据项”喂给同一个函数并发射每一个中间结果。它与reduce的区别在于reduce只发射最终结果scan发射全部累积过程值。Observable.just(5, 3, 8, 1, 7) .scan(0, (partialSum, x) - partialSum x) .subscribe(System.out::println); // prints: // 0 // 5 // 8 // 16 // 17 // 24scan的实现位于 Observable.javascan(BiFunctionT, T, T accumulator)为无种子版本对应测试见 ObservableScanTest.java。分组groupBygroupBy按照指定标准键选择器把源发射的数据项分组并以GroupedObservable或GroupedFlowable形式发射每一组。分组结果本身是可订阅的流可以继续应用操作符。仓库中定义了 GroupedObservable.java 与 GroupedFlowable.java 两种分组类型。ObservableString animals Observable.just( Tiger, Elephant, Cat, Chameleon, Frog, Fish, Turtle, Flamingo); animals.groupBy(animal - animal.charAt(0), String::toUpperCase) .concatMapSingle(Observable::toList) .subscribe(System.out::println); // prints: // [TIGER, TURTLE] // [ELEPHANT] // [CAT, CHAMELEON] // [FROG, FISH, FLAMINGO]示例按首字母分组并对组内元素应用String::toUpperCase值选择器再用concatMapSingle(Observable::toList)把每组收集成List输出。groupBy在 Observable.java 提供了多个重载仅键选择器、键选择器 值选择器、以及带StandardBufferedConfig配置的版本对应的 ObservableGroupByTest.java 覆盖了各种边界情况。批量聚合buffer 与 windowbuffer聚合成集合发射buffer把源发射的数据项收集到缓冲区中然后以集合默认List形式发射这些缓冲区。它的变体极为丰富按数量、按时间、按数量 步长、按边界信号、按开闭信号等。buffer(int count)按固定数量成批收集Observable.range(0, 10) .buffer(4) .subscribe((ListInteger buffer) - System.out.println(buffer)); // prints: // [0, 1, 2, 3] // [4, 5, 6, 7] // [8, 9]从源码看Observable.java 中buffer(int count)实际委托给buffer(count, count)Observable.java即“收集 count 个、跳过 count 个”的默认形态此外还有buffer(count, skip, SupplierU)自定义容器类型、buffer(timespan, timeskip, TimeUnit[, Scheduler])时间窗口版本、buffer(ObservableSourceB boundaryIndicator)边界信号版本等默认时间类变体使用Schedulers.computation()见 Observable.java。Flowable上同样提供了buffer(int count)Flowable.java且Flowable的buffer还支持背压感知。对应测试见 ObservableBufferTest.java。window切分成嵌套流window与buffer类似地切分数据但每个窗口本身是一个Observable或Flowable而不是集合。这意味着窗口内的数据项可以继续响应式地处理而不是一次性物化为列表。Observable.range(1, 10) // Create windows containing at most 2 items, and skip 3 items before starting a new window. .window(2, 3) .flatMapSingle(window - { return window.map(String::valueOf) .reduce(new StringJoiner(, , [, ]), StringJoiner::add); }) .subscribe(System.out::println); // prints: // [1, 2] // [4, 5] // [7, 8] // [10]示例中.window(2, 3)表示“每个窗口最多包含 2 个数据项每次跳过 3 个数据项再开启新窗口”因此得到[1,2]、[4,5]、[7,8]、[10]四个窗口。源码层面Observable.java 中window(long count)同样委托给window(count, count, bufferSize())并提供了时间窗口window(timespan, timeskip, TimeUnit[, Scheduler])、数量 时间混合window(timespan, unit, count[, restart])等大量变体与buffer一致时间类窗口默认使用Schedulers.computation()调度器。从源码看变换操作符的实现脉络所有变换操作符最终都会组装为internal/operators下的具体实现类并通过RxJavaPlugins.onAssembly(...)统一经过插件钩子参见 RxJavaPlugins.java 与 Observable.java 中各操作符的返回语句。以Observable为例可重点关注Observable.javamap、cast、bufferL4711 起、groupByL7826 起、flatMapL7278 起、concatMapL5550 起、switchMapL11168 起、scanL10180、windowL13313 起等全部声明于此。Flowable.javaFlowable版本的同类操作符如buffer(int count)L5275并带有背压语义。src/main/java/io/reactivex/rxjava4/internal/operators/observable/与.../operators/flowable/各操作符的具体实现类如ObservableBuffer、ObservableFlatMap、ObservableConcatMap、ObservableSwitchMap、ObservableGroupBy、ObservableWindow等。测试目录src/test/java/io/reactivex/rxjava4/internal/operators/observable/与.../operators/flowable/每个操作符都有对应的专项测试如 ObservableFlatMapTest.java、ObservableConcatMapTest.java、ObservableWindowTests.java 等是验证语义尤其是并发顺序、错误延迟行为的最佳参考资料。操作符选型速查需求推荐操作符每个数据项应用函数得到新值map每个数据项做类型转换cast或ofType先过滤再转换每个数据项展开成反应式源允许并发交错flatMap每个数据项展开成同步集合flatMapIterable每个数据项展开成源要求严格顺序输出concatMap/concatMapEager执行并行、输出串行只关心副作用完成、不需要结果值flatMapCompletable/concatMapCompletable只保留最新一次的结果、丢弃过期源switchMap需要发射每一次累积中间结果scan按键把数据流分组再分别处理groupBy按数量/时间/边界把数据聚合成集合buffer按数量/时间/边界把数据切分成嵌套流window希望错误不中断处理、最后统一上报各操作符的...DelayError变体总结变换操作符是 RxJava 数据管道的核心积木。理解flatMap并发合并、顺序不定、concatMap串行拼接、顺序严格、switchMap只取最新三者之间的差异以及...DelayError变体对错误时序的控制是写出正确响应式程序的关键。建议动手运行本文全部示例并结合 docs/Transforming-Observables.md、src/main/java/io/reactivex/rxjava4/internal/operators/下的实现类与对应的测试用例反复印证即可在实战中游刃有余地组合这些操作符。【免费下载链接】RxJavaRxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.项目地址: https://gitcode.com/gh_mirrors/rx/RxJava创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考