• 后端

导读

FlatMap 是响应式编程中最核心的“展平（flatten）”算子之一：它把上游 Observable 发出的每一个条目再次映射为一个 Observable，然后把所有这些内层 Observable 发出的条目合并、拍平成一条单一的数据流，供下游统一消费。在 RxGo 中，FlatMap 常用于“一对多”的级联变换场景——例如根据一个 ID 查询出多条关联记录、把事件流展开成细粒度事件序列、或将异步子任务的结果重新汇入主流水线。读完本文，你将掌握 FlatMap 的完整签名与基本用法、它与 Map 的本质区别、底层源码实现与错误处理语义，以及WithPool、WithCPUPool、WithContext、WithErrorStrategy等配套选项的实战配置方案。

FlatMap 是什么：定义与适用场景

按照 RxGo 官方文档 doc/flatmap.md 的定义：

FlatMap 将 Observable 发出的条目变换为 Observable，然后将这些 Observable 的发射结果展平为单个 Observable。

也就是说，FlatMap 承担两层职责：

1. 映射（Map 层）：对上游的每个条目Item应用一个函数，函数返回一个新的 Observable（每个条目对应一个“子流”）；
2. 展平（Flatten 层）：订阅这些子流，把它们的发射结果按顺序汇入同一条输出流，最终对外表现为“一个 Observable 连续发射一串条目”。

典型应用场景包括：

• 将“用户列表”展平为“每个用户的所有订单”；
• 将“目录列表”展平为“目录下所有文件”；
• 将“事件批次”展开为“单个事件流”；
• 在异步流水线中，为每个上游条目启动一次 IO 或外部调用，并把结果重新汇入主数据流。

在 RxGo 的算子分类中，FlatMap 属于“Transforming Observables”（变换类算子），与Map、GroupBy、Scan等并列，详见 README.md#transforming-observables 中的算子清单。

函数签名与核心类型

FlatMap 是Observable接口的成员方法，接口声明位于 observable.go：

FlatMap(apply ItemToObservable, opts ...Option) Observable

其中核心回调类型ItemToObservable定义在 types.go：

// ItemToObservable defines a function that computes an observable from an item. ItemToObservable func(Item) Observable

注意两点：

• 回调接收的是Item而不是裸值：Item是 RxGo 对“值或错误”的统一包装（结构体定义见 item.go），通过item.V取值、item.E取错误、item.Error()判断是否包含错误。这意味着映射函数可以感知并处理上游的错误条目。
• 支持柯里化选项：opts ...Option是可变参数，允许你在调用 FlatMap 时直接传入 doc/options.md 中定义的各类功能选项（缓冲、上下文、错误策略、并行池等），从而在不改动回调逻辑的前提下调整算子的执行行为。

基础用法：官方示例与输出解读

doc/flatmap.md给出了最小可用示例：用Just创建静态 Observable，FlatMap 把每个整数同时放大 10 倍和 100 倍，形成两条子流后拍平输出。

observable := rxgo.Just(1, 2, 3)().FlatMap(func(i rxgo.Item) rxgo.Observable { return rxgo.Just(i.V.(int) * 10, i.V.(int) * 100)() })

输出（顺序与上游条目一致）：

10 100 20 200 30 300

这里rxgo.Just(1, 2, 3)()是柯里化调用：第一组参数是条目列表，第二组参数是选项（Just的工厂实现见 factory.go）。FlatMap 回调中对i.V做了.(int)类型断言，这是 RxGo 基于interface{}的取值方式——上游条目类型必须与断言类型一致，否则会 panic，实践中应配合类型保护或错误处理使用。

从输出可以确认默认的顺序展平语义：对条目 1 生成的子流（10、100）全部发完，才轮到条目 2 的子流（20、200），最后是条目 3（30、300）。这正是 FlatMap 与 Merge 等并发合并算子的关键差异——FlatMap 默认严格按上游顺序串行处理。

源码级实现原理：FlatMap 到底做了什么

FlatMap 的实现位于 observable_operator.go，其核心是一个内部闭包f，由customObservableOperator（定义见 observable.go）包装成新的 Observable。逐段拆解如下：

f := func(ctx context.Context, next chan Item, option Option, opts ...Option) { defer close(next) observe := o.Observe(opts...) for { select { case <-ctx.Done(): return case item, ok := <-observe: if !ok { return } observe2 := apply(item).Observe(opts...) loop2: for { select { case <-ctx.Done(): return case item, ok := <-observe2: if !ok { break loop2 } if item.Error() { item.SendContext(ctx, next) if option.getErrorStrategy() == StopOnError { return } } else { if !item.SendContext(ctx, next) { return } } } } } } }

其工作流程可以归纳为三个层次：

1. 订阅上游：o.Observe(opts...)订阅源 Observable，并持续从通道中取出条目；ctx.Done()用于响应取消信号，实现优雅退出。
2. 逐条映射为子流：对取出的每个item，调用apply(item)得到一个新的 Observable，随即通过apply(item).Observe(opts...)订阅它（observe2）。
3. 内层循环展平：loop2循环逐个转发子流的条目到输出通道next，直到该子流关闭（!ok时break loop2），再回到外层循环处理下一个上游条目。

错误处理语义就在内层循环中：

• 子流发出错误条目时（item.Error()为真），先把错误转发给下游，然后检查错误策略：
  • 默认StopOnError：立即return，整个 FlatMap 终止，输出通道关闭；
  • ContinueOnError：只转发错误但不中断，继续处理剩余条目。
• 正常条目通过SendContext(ctx, next)发送，若发送失败（例如上下文已取消）则直接返回。

这种“外层条目串行、内层子流逐条转发”的结构，从源码层面印证了默认的顺序性：只有当前子流耗尽后才会开始下一个。

与 Map 的区别：一对一 vs 一对多

很多初学者容易混淆 Map 与 FlatMap，二者的关键差异是返回值的维度：

• Map的回调Func把一个条目变换为一个值（func(ctx context.Context, i interface{}) (interface{}, error)），输出条目数与输入相同，是一对一变换；
• FlatMap的回调ItemToObservable把一个条目变换为一个 Observable，再由 FlatMap 把这个子流“拍扁”，因此一个输入条目可能产生 0 个、1 个或多个输出条目，是一对多变换。

实践中的一个朴素判断标准：如果映射函数内部需要“发起一次调用并返回结果”，用Map；如果映射函数内部需要“发起调用并产生一串结果（例如遍历、分批、级联查询）”，用FlatMap。在 README.md 的 “Transforming Observables” 一节中，Map与FlatMap作为两类不同粒度的变换算子并列存在，正体现了这种分工。

错误传播与容错：两个关键测试用例

仓库测试文件 observable_operator_test.go 中，FlatMap 相关的测试直接验证了上文的错误语义：

子流产生错误（默认 StopOnError）：Test_Observable_FlatMap_Error1中，当条目i.V == 2时，映射出的子流只发射errFoo，断言结果为HasItems(2, 10)+HasError(errFoo)——即错误发生前已转发的条目照常输出，错误本身被透传，后续条目不再处理。

obs := testObservable(ctx, 1, 2, 3).FlatMap(func(i Item) Observable { if i.V == 2 { return testObservable(ctx, errFoo) } return testObservable(ctx, i.V.(int)+1, i.V.(int)*10) }) Assert(ctx, t, obs, HasItems(2, 10), HasError(errFoo))

上游产生错误条目：Test_Observable_FlatMap_Error2中，源 Observable 直接包含错误条目errFoo，映射函数对错误条目返回子流0，断言结果为HasItems(2, 10, 0, 4, 30)+HasNoError()。这证明ItemToObservable的回调可以感知item.Error()并主动“吃掉”错误，把错误条目替换成正常值。

这两个用例共同说明：FlatMap 的错误处理是可编程的——错误既可以被透传并终止（默认策略），也可以在回调内部被捕获并转化为正常数据。

并行化：WithPool 与 WithCPUPool

默认情况下 FlatMap 是串行的：一个 goroutine 顺序处理上游条目。对于 IO 密集、且每个条目需要独立执行子任务的场景，可以用WithPool/WithCPUPool选项把 FlatMap 变成并行算子：

// 指定 32 个 goroutine 并行执行映射与展平 observable := rxgo.Just(1, 2, 3)().FlatMap( func(i rxgo.Item) rxgo.Observable { return rxgo.Just(i.V.(int) * 10, i.V.(int) * 100)() }, rxgo.WithPool(32), )

• WithPool(pool int)：创建指定数量（pool）的并发 goroutine，定义见 options.go；
• WithCPUPool()：等价于WithPool(runtime.NumCPU())，按逻辑 CPU 数量自动决定并发度，定义见 options.go。

测试Test_Observable_FlatMap_Parallel（observable_operator_test.go）给出了并行模式下的断言方式：HasItemsNoOrder(2, 10, 3, 20, 4, 30)明确说明——并行后不再保证输出顺序，因此下游断言必须使用“无序集合”语义。

obs := testObservable(ctx, 1, 2, 3).FlatMap(func(i Item) Observable { return testObservable(ctx, i.V.(int)+1, i.V.(int)*10) }, WithCPUPool()) Assert(ctx, t, obs, HasItemsNoOrder(2, 10, 3, 20, 4, 30))

重要提醒：如果你的下游逻辑依赖条目顺序（例如按 ID 序列化输出），并行 FlatMap 会破坏原有顺序。此时应配合rxgo.Serialize选项强制串行化输出，或干脆保持默认串行模式。相关说明见 doc/serialize.md。

FlatMap 支持的选项全解

doc/flatmap.md列出 FlatMap 可用的全部选项，完整说明位于 doc/options.md，结合 options.go 的源码实现整理如下：

WithBufferedChannel(capacity int)

配置算子输出通道的容量（默认无缓冲）。例如WithBufferedChannel(1)创建容量为 1 的缓冲通道，可在子流生产速度快于下游消费速度时减少阻塞：

observable := rxgo.Just(1, 2, 3)().FlatMap(apply, rxgo.WithBufferedChannel(10))

源码中该选项写入options.buffer（见 options.go），在customObservableOperator创建next通道时生效。

WithContext(ctx)

传入context.Context，Observable 会监听其 Done 信号，一旦取消就关闭自身。这在超时控制、服务优雅停机场景中尤为关键：

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() observable := rxgo.Just(1, 2, 3)().FlatMap(apply, rxgo.WithContext(ctx))

源码实现中，customObservableOperator通过option.buildContext(parent)把传入的 ctx 与父级 ctx 合并（见 observable.go），FlatMap 内部闭包的select分支会在ctx.Done()时立即退出并关闭输出通道。

WithObservationStrategy(strategy)

设置观测策略（见 options.go）：

• rxgo.Lazy（默认）：只有 Observer 开始订阅时才消费并处理上游条目；
• rxgo.Eager：Observable 一创建就立即处理，即使没有订阅者。

// Eager 模式：创建即处理 observable := rxgo.Just(1, 2, 3)().FlatMap(apply, rxgo.WithObservationStrategy(rxgo.Eager))

WithErrorStrategy(strategy)

设置错误策略（见 options.go）：

• rxgo.StopOnError（默认）：产生错误即停止处理；
• rxgo.ContinueOnError：产生错误后继续处理剩余条目。

该策略会被传播到父 Observable（README 与 doc/options.md 均明确说明这一点）。对 FlatMap 而言，它直接决定上文实现中内层循环遇到错误时是return还是继续loop2。

observable := rxgo.Just(1, 2, 3)().FlatMap(apply, rxgo.WithErrorStrategy(rxgo.ContinueOnError))

WithPool(pool int) / WithCPUPool()

将算子转换为并行算子：WithPool(8)创建 8 个 goroutine；WithCPUPool()按runtime.NumCPU()创建。如前文所述，并行会牺牲输出顺序，适合 IO 密集的独立子任务。

WithPublishStrategy()

把普通 Observable 转换为 Connectable Observable（options.go）：Observable 在显式调用Connect()之前不会开始发射条目，所有 Observer 订阅完成后统一启动，且每个 Observer 都会收到一份条目拷贝。创建与连接方式如下：

ch := make(chan rxgo.Item) go func() { ch <- rxgo.Of(1) ch <- rxgo.Of(2) close(ch) }() observable := rxgo.FromChannel(ch, rxgo.WithPublishStrategy()) observable.FlatMap(apply) observable.Connect()

该策略同样会传播到父 Observable。Connectable Observable 的完整行为说明见 README.md#connectable-observable。

完整实战示例：级联查询并汇入单一数据流

综合以上所有要点，下面给出一个可直接运行的完整示例：源 Observable 发射一组订单 ID，FlatMap 为每个 ID 查询出多条明细记录，并以WithCPUPool并行执行查询，最终用ForEach消费拍平后的数据流。

package main import ( "context" "fmt" "github.com/reactivex/rxgo/v2" ) // findDetails 模拟按订单 ID 查询明细（返回 2 条记录） func findDetails(orderID int) rxgo.Observable { return rxgo.Just( fmt.Sprintf("order-%d-item-1", orderID), fmt.Sprintf("order-%d-item-2", orderID), )() } func main() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() // 上游：三个订单 ID source := rxgo.Just(1001, 1002, 1003)() // FlatMap：每个 ID 展开为两条明细，并行查询，输出通道缓冲 4 flat := source.FlatMap( func(i rxgo.Item) rxgo.Observable { return findDetails(i.V.(int)) }, rxgo.WithCPUPool(), rxgo.WithBufferedChannel(4), rxgo.WithContext(ctx), rxgo.WithErrorStrategy(rxgo.StopOnError), ) // 消费拍平后的数据流（结果无序，因为启用了并行） <-flat.ForEach( func(v interface{}) { fmt.Println("received:", v) }, func(err error) { fmt.Printf("error: %v\n", err) }, func() { fmt.Println("observable is closed") }, ) }

运行后你会看到 6 条order-xxx-item-y记录，顺序因并行而不固定，但observable is closed一定在最后打印，表示 FlatMap 及其所有子流已全部完成。若想恢复严格顺序，去掉rxgo.WithCPUPool()（或改为WithPool(1)）即可。

使用注意事项小结

1. 类型断言安全：回调中i.V.(int)之类的断言在类型不匹配时会 panic，建议在回调入口用类型保护或错误处理兜底。
2. 顺序 vs 性能的取舍：默认串行保证顺序；WithPool/WithCPUPool提升吞吐但打乱顺序，需要顺序时配合Serialize。
3. 错误策略影响范围：WithErrorStrategy会传播到父 Observable，配置前确认整条流水线的错误处理预期。
4. 子流的生命周期：每个上游条目对应的子流必须能够正常关闭，FlatMap 依赖observe2通道关闭（!ok）来判断子流结束；若子流永不关闭，展平将无限等待。
5. 资源清理：使用WithContext传入可取消的 context，并在不需要时调用cancel()，确保 FlatMap 内部的两个select分支能及时响应并关闭输出通道，避免 goroutine 泄漏（仓库测试全部以goleak.VerifyNone校验无泄漏，见 observable_operator_test.go）。

延伸阅读

• 算子选项完整说明：doc/options.md
• 相同变换家族的算子：doc/map.md、doc/groupby.md
• 展平语义的对比实现：doc/merge.md（无序合并）
• FlatMap 接口声明与实现：observable.go、observable_operator.go
• 测试用例与错误语义验证：observable_operator_test.go
• 安装方式：go get -u github.com/reactivex/rxgo/v2

• 后端