如何仅在flowA非空时合并Kotlin Flow并按需启停flowB

你需要的核心是根据FlowA的当前值动态启停FlowB——仅当FlowA发射非null值时订阅FlowB，发射null时立即取消FlowB的订阅，后续FlowA再次发射非null值时重新订阅，以此避免FlowB无意义运行。

用标准Flow算子实现 #

可以通过flatMapLatest结合emptyFlow()直接实现需求，完全不需要自定义算子：

flowA
    .flatMapLatest { a ->
        // 当a为null时，返回空流，相当于取消FlowB的订阅
        if (a != null) {
            // 非null时，将FlowB的每个值与当前的a配对
            flowB.map { b -> a to b }
        } else {
            emptyFlow()
        }
    }
    .collect { (a, b) ->
        Log.d(TAG, "$a -> $b")
    }

原理说明 #

• flatMapLatest的特性是：每当上游（FlowA）发射新值时，会立即取消之前订阅的子流（即之前的FlowB实例），然后订阅新的子流。
• 当FlowA发射null时，子流是emptyFlow()，此时没有有效流被订阅，FlowB不会运行。
• 当FlowA发射非null值时，子流是FlowB的映射流，此时FlowB才会被激活并发射值，直到FlowA下次发射新值（或null）。

为什么之前的方法不适用：

• takeWhile遇到null会直接终止整个FlowA，无法接收后续的非null更新。
• filterNotNull只是过滤掉FlowA的null值，但combine会一直保持FlowB的订阅，导致FlowB持续运行。

封装自定义算子（可选） #

如果需要复用这个逻辑，可以封装成自定义扩展函数，比如命名为combineWhenNotNull：

// 自定义扩展函数
fun <A : Any, B, R> Flow<A?>.combineWhenNotNull(
    flowB: Flow<B>,
    transform: suspend (A, B) -> R
): Flow<R> = flatMapLatest { a ->
    if (a != null) {
        flowB.map { b -> transform(a, b) }
    } else {
        emptyFlow()
    }
}

使用时更简洁：

flowA
    .combineWhenNotNull(flowB) { a, b ->
        Log.d(TAG, "$a -> $b")
    }
    .collect()

额外注意事项 #

• 如果FlowB是冷流：每次flatMapLatest订阅时都会重新启动FlowB，正好符合“需要才运行”的需求。
• 如果FlowB是热流：可以根据需求添加shareIn等算子控制共享行为，但要确保在不需要时能正确取消订阅，避免资源浪费。

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