一对一转换

map

将 Flow 中发出的每个元素转换为新元素，实现一对一映射。

适用场景：需要将每个项转换为不同类型或格式，例如在 ViewModel 中将 Date 对象格式化为可显示的 String。

filter

仅允许满足给定条件的元素通过。

适用场景：根据条件丢弃不需要的值，例如过滤空搜索词或无效数据。

take

限制性操作符，仅发出 Flow 的前 n 个元素，随后取消 Flow 执行。

适用场景：只需要有限数量的项，例如分页加载，或从一系列操作中取第一个有效结果。

累积值

reduce

终端操作符，从 Flow 的第一个元素开始累积值，并对当前累加器与每个元素执行操作。若 Flow 为空则抛出异常。

fold

与 reduce 类似，但允许指定初始累加值。即使流为空，它也能返回初始值，避免了空流抛出的异常风险，适用于需要默认值的计算场景。

发出多个值

transform

这是一个高度灵活的操作符，允许为每个输入元素发出零个、一个或多个值，从而对输出流拥有更强的控制力。

fun main() = runBlocking {
    (1..2).asFlow()
        .transform {
            emit("Item: $it")
            if (it % 2 != 0) {
                emit("...is an odd number")
            }
            emit("Square: ${it * it}")
        }.collect { println(it) }
}

// output:
// Item: 1
// ...is an odd number
// Square: 1
// Item: 2
// Square: 4

适用场景：执行复杂转换、引入副作用（如日志记录），或根据单个输入有条件地发出多个值。

扁平化嵌套

flatMapConcat

该操作符将每个元素转换为一个 Flow，然后依次连接这些 Flow。其核心特性在于顺序执行：只有当前一个 Flow 完成后，下一个才开始处理。

fun getNumbersFlow(id: Int): Flow = flow {
    delay(100)
    emit("First-$id")
    delay(100)
    emit("Second-$id")
}

fun main() = runBlocking {
    (1..2).asFlow().flatMapConcat { id -> getNumbersFlow(id) }.collect { println(it) }
}

// output:
// First-1
// Second-1
// First-2
// Second-2

仔细观察示例，第一个流的每个数字都会参与到第二个流的处理过程中。

适用场景：适用于对顺序敏感的操作，例如依次上传多个文件，或执行存在依赖关系的网络请求。

flatMapMerge

上下文与缓冲

flowOn

该操作符用于更改执行上游 Flow 的 CoroutineContext，它是 Flow 中切换调度器的标准且正确的方式。通过指定特定的协程上下文，开发者可以灵活地控制流在哪个线程或调度器上运行，从而实现计算密集型任务与 I/O 密集型任务的分离，优化应用性能。

buffer

buffer 操作符通过解耦生产者与消费者来实现并发执行。其核心机制是：生产者将生成的项放入缓冲区，而消费者则从缓冲区中取出并处理这些项。这种机制允许生产者和消费者以不同的速率运行，互不阻塞。

得益于 buffer 机制，整个数据的收集时间通常会小于 (200 + 300) * 3。这是因为缓冲允许生产者在消费者处理当前项时继续生成后续项，从而重叠了生成与处理的时间窗口。使用场景主要适用于生产者与消费者处理速度不一致的情况，通过引入缓冲来平衡负载，显著提升整体性能。

conflate

conflate 是一种特殊的缓冲形式。当收集器处理速度较慢时，它会丢弃中间产生的值，确保收集器始终只获取最新的值。与 buffer 不同，conflate 不保留历史数据，而是直接覆盖，从而减少内存占用和处理延迟。

适用场景包括 UI 更新中无需显示中间状态的情况，例如股票行情实时刷新或 GPS 位置更新。在这些场景中，用户通常只关心最新的状态，中间的历史值既无意义又浪费资源，因此 conflate 是理想选择。

collectLatest

作为终端操作符，collectLatest 的行为特点是：当新值发出时，它会立即取消对前一个值的收集逻辑，并开始处理新值。这意味着如果前一个收集过程耗时较长，它会被强制中断。

注意看这里的结果，Finished collecting 只收集了最后一次的值，一定要注意这个特性。 这一行为确保了最终输出的总是最新的数据。适用场景包括某项操作耗时较长，且应在新项到达时被取消的情况，例如将用户输入保存到数据库。如果用户快速连续输入，我们可能只希望保存最后一次输入，而非每一次击键。

合并

zip

错误与完成处理

catch

该操作符用于捕获上游 Flow 中发生的异常，这里的上游指的是 catch 之前的操作符链。需要注意的是，它不会捕获下游收集器中抛出的异常。通过 catch，开发者可以优雅地处理错误、提供默认值或记录失败信息，从而避免整个流因未处理的异常而终止。

onCompletion

此操作符在 Flow 完成时执行指定的回调操作，无论流是正常结束还是因异常而终止。在成功完成的情况下，传递给回调的 cause 参数为 null，这允许开发者在流结束时执行清理工作或记录日志。

retryWhen

当流发生异常时，retryWhen 允许根据特定的谓词逻辑决定是否进行重试。该谓词接收两个关键参数以辅助决策：首先是 cause，即导致 Flow 失败的异常对象，其类型为 Throwable；其次是 attempt，表示当前是第几次重试尝试，计数从 0 开始。通过灵活配置这些参数，可以实现复杂的重试策略，如指数退避或最大重试次数限制，从而增强流处理的健壮性。

工具与副作用

onEach

对 Flow 中每个元素执行指定操作，但不修改元素本身。

suspend fun main() {
    (1..3).asFlow()
        .onEach { println("About to process $it") }
        .map { it * it }
        .collect { println("Processed value: $it") }
}

// output:
// About to process 1
// Processed value: 1
// About to process 2
// Processed value: 4
// About to process 3
// Processed value: 9

适用场景：用于日志、调试或埋点等副作用场景，在不改变数据的前提下观察 Flow。

debounce

过滤在指定超时内被新值取代的值，仅发出“突发”中的最后一个值。

suspend fun main() {
    flow {
        emit(1)
        delay(90)
        emit(2)
        delay(90)
        emit(3)
        delay(500)
        emit(4)
        delay(90)
        emit(5)
    }.debounce(100).collect { println(it) } 
}
// output:
// 3
// 5

适用场景：处理快速用户输入（如搜索框），避免每次按键都触发 API 请求。

distinctUntilChanged

抑制与前一个值相同的重复发射。

suspend fun main() {
    flowOf(1, 1, 2, 2, 1, 3)
        .distinctUntilChanged()
        .collect { println(it) } 
}

// output:
// 1
// 2
// 1
// 3

适用场景：防止 UI 因状态未变而进行不必要的重组或更新。

核心决策指南

数据转换与累积

数据转换操作符用于修改流中的值，常见的包括 map、filter 以及 distinctUntilChanged。若需跟踪状态或计算持续结果，可以使用累积值操作符，例如 scan、runningReduce、fold 和 reduce。

动态发射与嵌套处理

当需要从单一输入发出多个值以自定义发射逻辑时，可使用 transform。对于处理嵌套或动态的 Flow，应根据内部 Flow 行为进行选择：flatMapConcat 用于顺序执行，flatMapMerge 用于并发执行，而 flatMapLatest 则负责取消旧任务并保留最新结果。

性能优化与流合并

为了优化收集效率与响应性，性能与背压控制操作符包括 buffer、conflate、collectLatest 以及 flowOn。在合并多个流以组合数据源时，zip 按顺序配对，combine 组合最新值，而 merge 则实现交错合并。

错误处理与调试

应对异常和生命周期事件的错误处理与完成逻辑操作符包括 catch、onCompletion 和 retryWhen。若需插入日志或副作用操作，可使用调试与副作用操作符 onEach。针对高频发射场景，debounce 用于抑制过快的发射频率。

终端操作

最后，通过触发 Flow 执行来结束流。常用的终端操作符包括 collect 和 collectLatest，它们标志着数据流的最终消费点。

总结

本文系统梳理了 Kotlin Flow 的核心操作符，涵盖转换、过滤、聚合及错误处理等关键 API。通过 first 与 single 等示例，展示了如何高效构建响应式数据流。掌握 toList 与 reduce 能显著提升代码可读性。最后，fold 提供了最佳实践建议，帮助开发者在实际项目中灵活应用 Flow，实现更简洁、安全的异步编程逻辑。

fun main() = runBlocking {
    flowOf("Kotlin", "Flow")
        .map { "Length of '$it' is ${it.length}" }
        .collect { println(it) }
}

// output:
// Length of 'Kotlin' is 6
// Length of 'Flow' is 4

fun main() = runBlocking {
    (1..5).asFlow()
        .filter { it % 2 == 0 }
        .collect { println(it) }
}

// output:
// 2
// 4

fun main() = runBlocking {
    (1..10).asFlow()
        .take(3)
        .collect { println(it) }
}

// output:
// 1
// 2
// 3

fun main() = runBlocking {
    val sum = (1..3).asFlow().reduce { accumulator, value -> accumulator + value }
    println(sum)
}

// output:
// 6

fun main() = runBlocking {
    val sum = (1..3).asFlow().fold(100) { accumulator, value -> accumulator + value }
    println(sum)
}

// output:
// 106

fun main() = runBlocking {
    println("runningReduce:")
    (1..3).asFlow()
        .runningReduce { accumulator, value -> accumulator + value }
        .collect { println(it) } 

    println("scan:")
    (1..3).asFlow()
        .scan(0) { accumulator, value -> accumulator + value }
        .collect { println(it) }
}

// output:
// runningReduce:
// 1
// 3
// 6
// scan:
// 0
// 1
// 3
// 6

fun getNumbersFlow(id: Int): Flow = flow {
    delay(100)
    emit("First-$id")
    delay(100)
    emit("Second-$id")
}

suspend fun main() {
    (1..2).asFlow()
        .flatMapMerge { id -> getNumbersFlow(id) }
        .collect { println(it) }
}

// output:
// First-1
// First-2
// Second-2
// Second-1

val searchQuery = flowOf("search", "search with new term").onEach { delay(200) }

fun searchApi(query: String): Flow = flow {
    emit("Searching for '$query'...")
    delay(500) // 模拟网络延迟
    emit("Results for '$query'")
}

suspend fun main() {
    searchQuery
        .flatMapLatest { query -> searchApi(query) }
        .collect { println(it) }
}

// output:
// Searching for 'search'...
// Searching for 'search with new term'...
// Results for 'search with new term'

fun heavyWork(): Flow<Int> = flow {
    println("Starting heavy work on ${Thread.currentThread().name}")
    for (i in 1..3) {
        // Simulate CPU-intensive work
        Thread.sleep(100)
        emit(i)
    }
}

fun main() = runBlocking {
    heavyWork()
        .flowOn(Dispatchers.IO) // Upstream runs on IO dispatcher
        .collect {
            println("Collected $it on ${Thread.currentThread().name}")
        }
    // Downstream runs on the collector's context (e.g., Main)
}

// output:
// Starting heavy work on DefaultDispatcher-worker-1
// Collected 1 on main
// Collected 2 on main
// Collected 3 on main

suspend fun main() {
    val time = measureTimeMillis {
        flow {
            for (i in 1..3) {
                delay(200) // Simulate slow emission
                emit(i)
            }
        }
        .buffer() // With buffer, the total time is closer to the slow collector's time
        .collect {
            delay(300) // Simulate slow collection
            println(it)
        }
    }
    println("Collected in $time ms")
}

// output:
// 1
// 2
// 3
// Collected in 1172 ms

suspend fun main() {
    flow {
        for (i in 1..5) {
            delay(100)
            emit(i)
        }
    }
    .conflate()
    .collect { value ->
        println("Started processing $value")
        delay(300)
        println("Finished processing $value")
    }
}

// output:
// Started processing 1
// Finished processing 1
// Started processing 3
// Finished processing 3
// Started processing 5
// Finished processing 5

suspend fun main() {
    (1..3).asFlow()
        .onEach { delay(100) }
        .collectLatest { value ->
            println("Collecting $value")
            delay(300)
            println("Finished collecting $value")
        }
}

// output:
// Collecting 1
// Collecting 2
// Collecting 3
// Finished collecting 3

suspend fun main() {
    val flowA = (1..3).asFlow()
    val flowB = flowOf("A", "B", "C", "D")

    flowA.zip(flowB) { number, letter -> "$number$letter" }
        .collect { println(it) } 
}

// output:
// 1A
// 2B
// 3C

suspend fun main() {
    val flowA = (1..3).asFlow().onEach { delay(100) }
    val flowB = flowOf("A", "B").onEach { delay(150) }

    flowA.combine(flowB) { number, letter -> "$number$letter" }
        .collect { println(it) }
}

// output:
// 1A
// 2A
// 3A
// 3B

suspend fun main() {
    val flowA = flowOf("A1", "A2").onEach { delay(100) }
    val flowB = flowOf("B1", "B2").onEach { delay(50) }

    merge(flowA, flowB)
        .collect { println(it) }
}

// output:
// B1
// A1
// B2
// A2

suspend fun main() {
    flow {
        emit(1)
        throw RuntimeException("Error!")
    }
    .catch { e ->
        println("Caught: ${e.message}")
        emit(-1) // Emit a fallback value
    }
    .collect { println(it) } // Emits 1, then -1
}

// output:
// 1
// Caught: Error!
// -1

suspend fun main() {
    (1..3).asFlow()
        .onCompletion { cause ->
            if (cause != null) println("Flow completed with error")
            else println("Flow completed successfully")
        }
        .collect { println(it) }
}

// output:
// 1
// 2
// 3
// Flow completed successfully

suspend fun main() {
    var attemptCount = 0
    flow {
        emit(1)
        if (attemptCount < 2) {
            attemptCount++
            throw RuntimeException("Transient error")
        }
        emit(2)
    }
    .retryWhen { cause, attempt ->
        println("Attempt $attempt: Retrying due to ${cause.message}")
        delay(100) // Add a delay before retrying
        attempt < 2 // Retry up to 2 times
    }
    .catch { println("Caught final error: ${it.message}") }
    .collect { println(it) }
}

// output:
// 1
// Attempt 0: Retrying due to Transient error
// 1
// Attempt 1: Retrying due to Transient error
// 1
// 2