企业官网建设流程全解析

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。为忠实保留原文档语义，下文示例沿用文档写法，实际编译时按根包名替换即可。

变换操作符一览与可用性速查

以下表格汇总了全部变换操作符及其在五大反应式类型上的可用性（✓ 表示可用，✗ 表示不可用），方便你在选型时快速定位：

核心规律：凡是要求“每个数据项都产出多个结果”的操作（如map、cast、flatMap）基本都支持Maybe/Single；凡是涉及多数据项之间聚合、分组、串并关系的操作（如buffer、window、groupBy、scan）只存在于多值流Flowable/Observable上；Completable因不发射数据项，因此不提供任何变换操作符。

一对一映射：map 与 cast

map：逐项应用函数

map对源发射的每个数据项应用给定的io.reactivex.functions.Function，并发射函数计算结果。它是变换家族中最基础、最常用的操作符，在 Observable.java 附近的flatMap以及各操作符实现中随处可见其身影。

Observable.just(1, 2, 3) .map(x -> x * x) .subscribe(System.out::println); // prints: // 1 // 4 // 9

cast：按类型转换

cast把源发射的每个数据项强制转换为指定类型后发射。它本质上等价于map内部的类型转换，但在类型不匹配时会抛ClassCastException。在 Observable.java 中，cast(Class<U> clazz)被定义为一个泛型方法，返回Observable<U>；Observable还有一个便捷的ofType组合（见 Observable.java），它先按类型过滤再cast。

Observable<Number> 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。它适合“对每个数据项执行副作用操作（如写库、发消息）但不需要结果值”的场景。

Observable<Integer> 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 completed

flatMapIterable：展开同步集合

当映射函数返回的是同步的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 // C

flatMapMaybe：与 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 // onComplete

flatMapSingle：与 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会发出错误通知而不是静默完成：

Maybe<Object> emptySource = Maybe.empty(); Single<Object> 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发射的单个数据项交给映射函数，展开成一个ObservableSource（flatMapObservable）或org.reactivestreams.Publisher（flatMapPublisher），分别返回Observable与Flowable。适合“一个值拆成多个值”的场景，例如把 CSV 字符串拆成多行。

Single<String> source = Single.just("Kirk, Spock, Chekov, Sulu"); Observable<String> 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: Sulu

Single<String> source = Single.just("Kirk, Spock, Chekov, Sulu"); Flowable<String> 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: Sulu

flatMapSingleElement：Maybe 版的不报错展开

flatMapSingleElement仅存在于Maybe上：映射函数返回io.reactivex.SingleSource，若源Maybe有值则发射该Single的结果，若源Maybe为空则直接完成，绝不报错。

Maybe<Integer> source = Maybe.just(-42); Maybe<Integer> result = source.flatMapSingleElement(x -> { return Single.just(Math.abs(x)); }); result.subscribe(System.out::println); // prints 42

flattenAsFlowable 与 flattenAsObservable：把单个值展平为集合流

与flatMapObservable/flatMapPublisher类似，但映射函数返回同步的java.lang.Iterable，分别产出Flowable与Observable。

Single<Double> source = Single.just(2.0); Flowable<Double> 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.0

Single<Double> source = Single.just(2.0); Observable<Double> 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 AAABBBCCC

concatMapCompletable 与 concatMapCompletableDelayError

concatMapCompletable要求映射函数返回io.reactivex.CompletableSource，逐个订阅，全部完成后返回的Completable才完成：

Observable<Integer> 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与之相同，但延迟错误：某个源出错不会中断后续处理，所有源都终止后错误才被统一上报：

Observable<Integer> 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时，主源的错误会在当前内部源终止后立即上报。

Observable<Integer> 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 与 concatMapMaybeDelayError

concatMapMaybe要求映射函数返回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 与 concatMapSingleDelayError

concatMapSingle要求映射函数返回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.0

concatMapSingleDelayError则延迟所有源错误：

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

只取最新：switchMap

switchMap对源发射的每个数据项应用函数生成内部源，但只发射最近一个内部源的结果：每当新数据项到达，前一个内部源会被退订（取消订阅）。非常适合搜索框输入联想、竞态消除等“以最新输入为准”的场景。

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）。

累积扫描：scan

scan使用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 // 24

scan的实现位于 Observable.java（scan(BiFunction<T, T, T> accumulator)为无种子版本），对应测试见 ObservableScanTest.java。

分组：groupBy

groupBy按照指定标准（键选择器）把源发射的数据项分组，并以GroupedObservable（或GroupedFlowable）形式发射每一组。分组结果本身是可订阅的流，可以继续应用操作符。仓库中定义了 GroupedObservable.java 与 GroupedFlowable.java 两种分组类型。

Observable<String> 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 与 window

buffer：聚合成集合发射

buffer把源发射的数据项收集到缓冲区中，然后以集合（默认List）形式发射这些缓冲区。它的变体极为丰富：按数量、按时间、按数量 + 步长、按边界信号、按开闭信号等。

buffer(int count)按固定数量成批收集：

Observable.range(0, 10) .buffer(4) .subscribe((List<Integer> 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, Supplier<U>)自定义容器类型、buffer(timespan, timeskip, TimeUnit[, Scheduler])时间窗口版本、buffer(ObservableSource<B> 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.java：map、cast、buffer（L4711 起）、groupBy（L7826 起）、flatMap（L7278 起）、concatMap（L5550 起）、switchMap（L11168 起）、scan（L10180）、window（L13313 起）等全部声明于此。
• Flowable.java：Flowable版本的同类操作符，如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 等，是验证语义（尤其是并发顺序、错误延迟行为）的最佳参考资料。

操作符选型速查

总结

变换操作符是 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