亲爱的读者，欢迎来到《Kotlin Flow高级技巧：操作符函数大全》。在这篇文章中，我们将深入探讨Kotlin Flow中的各种操作符，并通过具体的示例代码，帮助你理解每个操作符的实际应用。这将是一次既丰富又详细的学习之旅。

转换操作符

1. map:

• 描述：转换流中的每个元素。
• 示例代码：

flowOf(1, 2, 3)

.map { it * it }

.collect { println(it) } // 输出: 1, 4, 9

2. filter:

• 描述：过滤流中的元素，只保留满足条件的元素。
• 示例代码：

flowOf(1, 2, 3)

.filter { it % 2 == 1 }

.collect { println(it) } // 输出: 1, 3

3. transform:

• 描述：提供了一种更灵活的方式来转换流中的元素。
• 示例代码：

flowOf("Kotlin", "Flow")

.transform { value ->

emit(value.length)

emit(value.uppercase())

}

.collect { println(it) } // 输出: 6, KOTLIN, 4, FLOW

4. transformLatest:

• 描述：类似于transform，但是在新值到来时会取消之前的转换操作。
• 示例代码：

(1..3).asFlow()

.transformLatest { value ->

emit("Processing $value")

delay(100) // 模拟耗时操作

emit("Done $value")

}

.collect { println(it) }

// 输出: Processing 1, Processing 2, Processing 3, Done 3

组合操作符

1. flatMapConcat:

• 描述：将流中的每个值转换为另一个流，并按顺序连接这些流。
• 适用场景：需要按照顺序处理每个元素时。
• 示例代码：

(1..3).asFlow()

.flatMapConcat { value ->

flowOf("Item: $value")

}

.collect { println(it) } // 输出: Item: 1, Item: 2, Item: 3

2. flatMapMerge:

• 描述：类似于flatMapConcat，但并行处理转换后的流。
• 适用场景：需要并行处理每个元素，无需按照特定顺序时。
• 示例代码：

(1..3).asFlow()

.flatMapMerge { value ->

flow {

emit("Start: $value")

delay(100) // 模拟耗时操作

emit("End: $value")

}

}

.collect { println(it) }

// 输出可能是乱序的，例如: Start: 1, Start: 2, End: 1, Start: 3, End: 2, End: 3

(1..3).asFlow()

.flatMapMerge { value ->

flow {

emit(value)

delay(100) // 模拟耗时操作

emit(value + 10)

}

}

.collect { println(it) } // 输出可能是乱序的，例如: 1, 2, 3, 11, 12, 13

3. zip:

• 描述：将两个流的相关元素配对，并应用某个函数。
• 适用场景：需要将两个流中相应位置的元素配对。
• 示例代码：

val flowA = flowOf("Hello")

val flowB = flowOf("World")

flowA.zip(flowB) { a, b -> "$a $b" }

.collect { println(it) } // 输出: Hello World

val flowA = flowOf("A", "B", "C")

val flowB = flowOf(1, 2, 3)

flowA.zip(flowB) { a, b -> "$a$b" }

.collect { println(it) } // 输出: A1, B2, C3

4. combine:

• 描述：将多个流组合在一起。
• 适用场景：需要组合两个流中的最新值。
• 示例代码：

val flowA = flowOf("Hello")

val flowB = flowOf("World")

flowA.combine(flowB) { a, b -> "$a $b" }

.collect { println(it) } // 输出: Hello World

val flowA = flowOf("A", "B", "C").onEach { delay(100) }

val flowB = flowOf(1, 2, 3).onEach { delay(150) }

flowA.combine(flowB) { a, b -> "$a$b" }

.collect { println(it) }

// 输出: A1, B1, B2, C2, C3

错误处理

1. catch:

• 描述：用于处理Flow中的异常。
• 示例代码：

flow {

emit(1)

throw RuntimeException("Error")

}

.catch { e -> emit("Caught $e") }

.collect { println(it) } // 输出: 1, Caught java.lang.RuntimeException: Error

2. retry:

• 描述：在出现错误时重试流的收集。
• 适用场景：在出现错误时重试固定次数。
• 示例代码：

var attempt = 0

flow {

if (++attempt < 3) {

throw Exception("Failed")

}

emit("Success")

}

.retry(2)

.catch { emit("Error") }

.collect { println(it) } // 输出: Success

3. retryWhen:

• 描述：在出现错误时重试流的收集。
• 适用场景：根据错误类型或重试次数自定义重试逻辑。
• 示例代码：

var attempt = 0

flow {

if (++attempt < 3) {

throw Exception("Failed")

}

emit("Success")

}

.retryWhen { cause, attempt -> cause is Exception && attempt < 2 }

.catch { emit("Error") }

.collect { println(it) } // 输出: Success

性能优化

1. conflate:

• 描述：在发射者比接收者快时跳过中间值，只处理最新值。
• 示例代码：

flow {

for (i in 1..3) {

delay(100) // 假设这是一项耗时操作

emit(i)

}

}

.conflate()

.collect { value ->

delay(300) // 模拟慢速消费者

println(value)

} // 输出: 1, 3 (跳过了2)

2. buffer:

• 描述：控制流的缓冲行为，提高处理效率。
• 示例代码：

flow {

for (i in 1..3) {

emit(i)

delay(100)

}

}

.buffer()

.collect { value ->

delay(300)

println(value)

} // 输出不会因为消费者的延迟而受阻

状态和生命周期操作符

1. stateIn:

• 描述：将Flow转换为状态流，提供最近一次的值。
• 示例代码：

val stateFlow = flowOf(1, 2, 3)

.stateIn(coroutineScope)

println(stateFlow.value) // 输出: 3

2. onStart 和 onCompletion:

• 描述：分别在流开始和结束时执行操作。
• 示例代码：

flowOf("Start", "Middle", "End")

.onStart { println("Flow starts") }

.onCompletion { println("Flow completes") }

.collect { println(it) }

// 输出: Flow starts, Start, Middle, End, Flow completes

特殊流构建器

1. flowOf 和 asFlow:

• 描述：用于创建Flow。
• 示例代码：

val flowA = flowOf(1, 2, 3)

val flowB = listOf(4, 5, 6).asFlow()

flowA.zip(flowB) { a, b -> a + b }

.collect { println(it) } // 输出: 5, 7, 9

2. emptyFlow:

• 描述：创建一个不发射任何元素的Flow。
• 示例代码：

emptyFlow<Int>()

.collect { println(it) } // 不输出任何东西

3. callbackFlow

• 描述：callbackFlow 用于将非协程代码（例如回调）集成到Flow中。它提供了一种安全的方式来发射值，直到流收集完成或被取消。
• 适用场景：适用于将现有的基于回调的API转换为响应式流。例如，集成传感器数据、位置更新或任何其他基于回调的API。
• 示例代码：

callbackFlow {

val callback = object : SomeCallback {

override fun onEvent(event: Event) {

trySend(event).isSuccess // 发射事件

}

}

registerCallback(callback)

awaitClose { unregisterCallback(callback) } // 在Flow被取消时解除回调

}

4. channelFlow:

• 描述：channelFlow 也用于在Flow中发射来自非协程代码的值。与 callbackFlow 不同，channelFlow 不具备自动完成的特性，需要显式关闭或完成。
• 适用场景：当你需要更多控制，或者当你的数据源本身就是多个独立操作的结果时。例如，从多个源收集数据或者在不同条件下发射值。
• 示例代码：

channelFlow {

val job1 = launch {

// 模拟一些异步操作

delay(100)

send("Result 1")

}

val job2 = launch {

// 另一个异步操作

delay(200)

send("Result 2")

}

job1.join() // 等待任务完成

job2.join()

close() // 手动关闭Flow

}

对比总结

• callbackFlow 通常用于处理基于回调的API，它会在流被取消时自动清理并关闭回调。
• channelFlow 则提供了更大的灵活性和控制权，适用于需要从多个独立源收集数据的场景。在 channelFlow 中，你需要手动管理关闭流的时机。

更多其他操作符

1. first 和 firstOrNull:

• 描述：从Flow中获取第一个元素，firstOrNull在没有元素时返回null。
• 示例代码：

val firstValue = flowOf(1, 2, 3).first() // 返回1

val firstOrNullValue = emptyFlow<Int>().firstOrNull() // 返回null

2. single 和 singleOrNull:

• 描述：确保Flow仅发射一个元素，否则抛出异常。singleOrNull在没有元素时返回null。
• 示例代码：

val singleValue = flowOf(1).single() // 返回1

val singleOrNullValue = emptyFlow<Int>().singleOrNull() // 返回null

3. distinctUntilChanged:

• 描述：仅当当前值与上一个值不同时，才发射该值。
• 示例代码：

flowOf(1, 1, 2, 2, 3)

.distinctUntilChanged()

.collect { println(it) } // 输出: 1, 2, 3

4. debounce:

• 描述：在指定时间内仅发射最新的元素。
• 适用场景：减少过快的数据发射频率，仅在指定时间内没有新值产生时，发射最新值。
• 示例代码

flowOf(1, 2, 3)

.debounce(100) // 延迟100毫秒

.collect { println(it) } // 输出: 3

5. sample:

• 描述：定期发射Flow中的最新值。
• 适用场景：定期采样流中的最新值。
• 示例代码：

flow {

emit(1)

delay(50)

emit(2)

delay(100)

emit(3)

}

.sample(100) // 每100毫秒采样一次

.collect { println(it) } // 输出: 1, 3