Kotlin Flow与RxJava的scan操作及共享流投递行为差异解析

差异核心和挂起Lambda的特性没有直接因果关系，本质是两套流框架的调度模型、元素传递实现逻辑的完全区别，逐帧对比执行时序即可清晰定位：

• RxJava版本采用全同步串行调用栈模型
  从subscribe触发开始，整个流的元素传递是沿操作符链同步递归调用的，不存在独立的异步执行单元。静态流Observable.just("item1", "item2")的发射逻辑，会在上游每次推完一个元素、下游整个处理栈执行完毕后，才会发射下一个元素，实际执行步骤为：
  1. 订阅启动，scan算子的初始值"*"被推到Subscriber，打印got *
  2. just流发射item1，经map转成pair，通过merge、scan计算得到累计值*_item1，推到Subscriber
  3. Subscriber执行pair关联的Lambda，调用shared.onNext("ITEM1")：因为PublishSubject的onNext是同步调用所有订阅者，这个ITEM1会立刻沿merge、scan的操作链计算出累计值*_item1_ITEM1，推到Subscriber执行，打印对应日志
  4. 上述整个调用栈全部返回后，just流才继续发射item2，重复上述流程，最终形成大写ITEM穿插在小写item之间的输出顺序
• 协程Flow版本采用并发收集+通道解耦模型
  kotlinx.coroutines内置的merge操作符，会为每个上游流启动独立的协程并发收集，上游发射的元素会先写入公共通道，再由单独的下游协程串行读取、经过后续算子（比如scan）处理。
  这种模型下，静态流flowOf("item1", "item2")的协程一旦启动，因为整个发射过程没有挂起点，会不等待下游处理节奏，一口气把item1、item2两个元素全部写入公共通道，协程随即结束。实际执行步骤为：
  1. merge启动三个协程：协程A负责收集shared流并转成pair写入通道，协程B负责收集静态item流并转成pair写入通道，协程C负责串行读取通道元素、经过scan处理后执行collect逻辑
  2. 协程B启动后，无挂起连续把item1、item2依次写入通道，随即结束，完全不等协程C的处理进度
  3. 协程C先处理scan初始值，打印got *
  4. 协程C从通道读取第一个元素item1，处理后执行对应Lambda，调用shared.emit("ITEM1")：此时协程A收到ITEM1，转成pair写入通道，通道内现有内容为[item2, ITEM1的pair]
  5. 协程C打印got *_item1，继续读通道，优先读到提前写入的item2，处理后执行对应Lambda调用emit("ITEM2")，协程A收到ITEM2写入通道，此时通道内内容为[ITEM1的pair, ITEM2的pair]
  6. 协程C打印got *_item1_item2，继续依次读取通道内的ITEM1、ITEM2对应元素，逐次打印后续三行，和实际观测到的输出完全一致

普遍的认知误区是认为RxJava依靠特殊缓冲实现了穿插输出，实际上恰恰相反：

• 没有独立的异步缓冲解耦上游发射和下游处理：所有算子的元素传递都是通过方法调用栈直接递归完成的，不存在协程版本里的公共通道把上游发射和下游处理拆开的设计
• PublishSubject的onNext方法没有额外的异步排队逻辑：一旦调用，会同步执行所有订阅者的处理逻辑，对应的元素会立刻走完整个操作符链的处理流程，不会被缓存到队列里等所有上游元素发完再统一处理
• 上游发射节奏和下游处理严格绑定：没有异步边界的情况下，上游的发射逻辑会等下游把当前元素的整个处理流程执行完、方法栈返回后，才会继续发射下一个元素，完全不存在协程版本里静态流一口气把所有元素塞到通道的情况

要实现完全一致的穿插输出效果，核心是打破协程版本中静态流无等待一口气发射所有元素的问题，把静态流的发射节奏和下游的处理节奏绑定，常用的有效调整方式有两种：

1. 给静态item流增加节奏控制，每发射一个元素主动让出调度权
   把原来的静态flowOf替换为带挂起点的流定义，通过yield()让发射协程暂停，等下游协程完成当前元素的全流程处理后，再发射下一个元素，示例代码：

   runBlocking {
     val shared = MutableSharedFlow<String>()
     merge(
       shared.map { it to suspend {} },
       flow {
         emit("item1")
         yield() // 主动让出调度权，等待其他协程完成当前流程
         emit("item2")
         yield()
       }.map { it to suspend { shared.emit(it.toUpperCase()) } }
     )
       .scan("*" to suspend { }) { accumulator, value ->
         (accumulator.first + "_" + value.first) to value.second
       }
       .collect {
         it.second()
         println("got ${it.first}")
       }
   }
   

   调整后最终的输出顺序和RxJava版本完全一致。
2. 移除通道解耦的merge实现，手动模拟RxJava的同步栈传递逻辑
   不使用内置的merge操作符，直接在单个flow的构建块里串行处理两个来源的元素，让元素的传递沿同步调用栈完成，示例代码：

   runBlocking {
     val shared = MutableSharedFlow<String>()
     val combinedFlow = flow {
       // 启动独立协程接收shared流的元素
       val sharedCollector = launch {
         shared.collect { value ->
           emit(value to suspend {})
         }
       }
       // 逐次发射静态item，每发射一次同步等待下游处理完成
       emit("item1" to suspend { shared.emit("ITEM1") })
       emit("item2" to suspend { shared.emit("ITEM2") })
       sharedCollector.cancel()
     }
     combinedFlow
       .scan("*" to suspend { }) { accumulator, value ->
         (accumulator.first + "_" + value.first) to value.second
       }
       .collect {
         it.second()
         println("got ${it.first}")
       }
   }
   

常见的无效调整方向：增大SharedFlow的缓冲容量、给流添加额外的buffer操作符，这些调整只会改变通道的排队顺序，无法解决上游提前把所有元素塞入通道的核心问题，无法对齐RxJava的输出效果。

内容的提问来源于stack exchange，提问作者dimsuz