企业官网建设流程全解析

Intermediate operators

消费上游流，返回新的下游流

Transforming operators

转换运算符转换上游流发出的值

repeat

suspend fun main() = withContext(Dispatchers.Default) { val flow = (0..4).asFlow().transform { value -> // Emits each value as many times as its value repeat(value) { emit(value) } } println(flow.toList()) }

[1, 2, 2, 3, 3, 3, 4, 4, 4, 4]

map

suspend fun main() = withContext(Dispatchers.Default) { // Multiplies each upstream value by four val flow = (0..4).asFlow().map { it * 4 } println(flow.toList()) }

[0, 4, 8, 12, 16]

filter

suspend fun main() = withContext(Dispatchers.Default) { // Emits only values where dividing by 3 leaves a remainder of 1 val flow = (0..10).asFlow().filter { it % 3 == 1 } println(flow.toList()) }

[1, 4, 7, 10]

mapNotNull

suspend fun main() = withContext(Dispatchers.Default) { // Converts each string to Double and skips values that can't be converted val flow = flowOf("1.2", "10", "11", "error", "0.000") .mapNotNull { it.toDoubleOrNull() } println(flow.toList()) }

[1.2, 10.0, 11.0, 0.0]

Filtering and size-limiting operators

过滤和大小限制运算符控制哪些值从流下游继续

distinctUntilChanged

suspend fun main() = withContext(Dispatchers.Default) { // Removes repeated consecutive values from the upstream flow val flow = flowOf(1, 2, 3, 3, 3, 4, 5, 5, 1).distinctUntilChanged() println(flow.toList()) }

[1, 2, 3, 4, 5, 1]

drop

suspend fun main() = withContext(Dispatchers.Default) { // Skips the first two values from the upstream flow val flow = flowOf(1, 2, 3, 4, 5).drop(2) println(flow.toList()) }

[3, 4, 5]

take

suspend fun main() = withContext(Dispatchers.Default) { // Collects only the first three values from the upstream flow val flow = (0..1000).asFlow().take(3) println(flow.toList()) }

[0, 1, 2]

Concurrent processing operators

同时运行上游流和下游收集，使用并发处理运算符引入缓冲区。缓冲区存储上游流已发出但收集器尚未处理的值。

buffer

默认情况，当缓冲区已满时，上游流暂停，当收集器释放空间时恢复

suspend fun main() = withContext(Dispatchers.Default) { flow { repeat(10) { emit(it) println("Emitted $it!") } } // Lets the upstream flow emit up to four values ahead of the collector .buffer(4) .collect { println("Processed $it!") delay(20.milliseconds) } }

Emitted 0! Processed 0! Emitted 1! Emitted 2! Emitted 3! Emitted 4! Processed 1! Emitted 5! Processed 2! Emitted 6! Processed 3! Emitted 7! Processed 4! Emitted 8! Processed 5! Emitted 9! Processed 6! Processed 7! Processed 8! Processed 9!

conflate

相当于buffer(1, onBufferOverflow = BufferOverflow.DROP_OLDEST)，即只处理最新值

若要取消已经开始的处理。请改用collectLatest

suspend fun main() = withContext(Dispatchers.Default) { flow { repeat(10) { emit(it) println("Emitted $it!") } }.conflate().collect { println("Processed $it!") delay(20.milliseconds) } }

Processed 0! Emitted 0! Emitted 1! Emitted 2! Emitted 3! Emitted 4! Emitted 5! Emitted 6! Emitted 7! Emitted 8! Emitted 9! Processed 9!

flowOn

改变调度器，在单独的协程中收集上游流，并在上游排放和下游收集之间使用缓冲区，如下将阻塞操作sleep转到Dispatchers.IO，而不是Dispatchers.Default中

fun awaitSensorSignal(): SensorSignal { Thread.sleep(10) val reading = round(Random.nextDouble(25.0, 100.0) * 100.0)/100.0 println("Measured $reading as the temperature") return SensorSignal(temperatureCelsius = reading) } data class SensorSignal( val temperatureCelsius: Double ) suspend fun sendLatestTemperature(temperatureCelsius: Double) { println("Starting to send $temperatureCelsius...") delay(50.milliseconds) println("Sent $temperatureCelsius.") } suspend fun main() = withContext(Dispatchers.Default) { val smartHomeTemperatureFlow = flow { while (true) { val signal = awaitSensorSignal() emit(signal.temperatureCelsius) println("Emitted $signal") } } // Runs the upstream flow in Dispatchers.IO .flowOn(Dispatchers.IO) // Keeps the newest buffered value and drops older ones .conflate() // Collects the first two values from the upstream flow .take(2) .collect { temperature -> println("Received $temperature!") sendLatestTemperature(temperature) } }

Measured 33.92 as the temperature Received 33.92! Starting to send 33.92... Emitted SensorSignal(temperatureCelsius=33.92) Measured 83.1 as the temperature Emitted SensorSignal(temperatureCelsius=83.1) Measured 40.42 as the temperature Emitted SensorSignal(temperatureCelsius=40.42) Measured 86.15 as the temperature Emitted SensorSignal(temperatureCelsius=86.15) Measured 43.33 as the temperature Emitted SensorSignal(temperatureCelsius=43.33) Sent 33.92. Received 43.33! Starting to send 43.33... Measured 66.58 as the temperature Emitted SensorSignal(temperatureCelsius=66.58) Measured 82.47 as the temperature Emitted SensorSignal(temperatureCelsius=82.47) Measured 59.08 as the temperature Emitted SensorSignal(temperatureCelsius=59.08) Measured 97.13 as the temperature Emitted SensorSignal(temperatureCelsius=97.13) Measured 68.86 as the temperature Emitted SensorSignal(temperatureCelsius=68.86) Sent 43.33. Measured 99.41 as the temperature

Combining operators

消耗多个上游流的值，并返回单个下游

zip

对来自两个上游流的值进行配对，其中一个上游流完成，结果流就完成了

suspend fun main() = withContext(Dispatchers.Default) { // Emits a ticker value every 100 milliseconds val tickerFlow = flow { while (true) { emit(Unit) delay(100.milliseconds) } } val start = TimeSource.Monotonic.markNow() tickerFlow // Combines each ticker emission with the next number .zip(flowOf(1, 2, 3)) { _, value -> value }.collect { println("${start.elapsedNow()}: received $it") } }

60.181144ms: received 1 166.750681ms: received 2 267.383307ms: received 3

combine

当任何上游流发出值时，它会使用每个上游流的最新值发出一个新值

enum class Theme { Dark, Light, } data class UiState( val messages: List<String>, val theme: Theme, ) val messagesFlow = MutableStateFlow( listOf( "Hello!", "Is anyone here?", ) ) val themeFlow = MutableStateFlow( Theme.Light ) // Combines the latest values from both upstream flows val uiStateFlow = combine(messagesFlow, themeFlow) { messages, theme -> UiState(messages, theme) } suspend fun main() { withContext(Dispatchers.Default) { // Uses UNDISPATCHED to subscribe before the first update happens val uiUpdateJob = launch(start = CoroutineStart.UNDISPATCHED) { uiStateFlow.collect { // Draws the UI println(it) } } messagesFlow.update { messages -> messages + "I'll be back!" } delay(100.milliseconds) themeFlow.value = Theme.Dark delay(100.milliseconds) uiUpdateJob.cancel() } }

UiState(messages=[Hello!, Is anyone here?], theme=Light) UiState(messages=[Hello!, Is anyone here?, I'll be back!], theme=Light) UiState(messages=[Hello!, Is anyone here?, I'll be back!], theme=Dark)

merge

从多个流中收集值并将它们的值发送到一个下游流中

interface UiEvent class ClickEvent: UiEvent class RightClickEvent: UiEvent suspend fun main() { withContext(Dispatchers.Default) { val clickFlow = MutableSharedFlow<ClickEvent>() val rightClickFlow = MutableSharedFlow<RightClickEvent>() coroutineScope { // Uses UNDISPATCHED to subscribe before the first update happens val collectJob = launch(start = CoroutineStart.UNDISPATCHED) { // Collects both upstream flows concurrently and emits their values downstream merge(clickFlow, rightClickFlow).collect { println("Observed an event: $it") } } clickFlow.emit(ClickEvent()) delay(100.milliseconds) clickFlow.emit(ClickEvent()) delay(100.milliseconds) rightClickFlow.emit(RightClickEvent()) delay(100.milliseconds) collectJob.cancel() } } }

Observed an event: ClickEvent@71e9d72c Observed an event: ClickEvent@28cb875f Observed an event: RightClickEvent@165c2d5f

Lifecycle operators

在流收集期间在特定点运行代码

onStart、onEach

onStart 在收集上游流之前运行代码，onEach在每个发出值之前运行代码

suspend fun main() { withContext(Dispatchers.Default) { flowOf("Page 1", "Page 2", "Page 3").onStart { println("Processing pages!") }.onEach { println("Emitted $it") }.collect { println("Collected $it") } } }

Processing pages! Emitted Page 1 Collected Page 1 Emitted Page 2 Collected Page 2 Emitted Page 3 Collected Page 3

onCompletion

在收集完成后运行代码

suspend fun main() { withContext(Dispatchers.Default) { flowOf("Page 1", "Page 2", "Page 3").onCompletion { println("Almost done...") // Emits an additional value after the upstream flow completes emit("Last Page!") }.collect { println("Collected $it") } } }

Collected Page 1 Collected Page 2 Collected Page 3 Almost done... Collected Last Page!

oEmpty

在上游流完成而不发出任何值时运行代码

suspend fun main() { withContext(Dispatchers.Default) { flowOf("Page 1", "Page 2", "Page 3").onEmpty { // Doesn't print anything, because the upstream flow emits values println("No pages to load!") }.collect() flowOf<Int>().onEmpty { println("No pages to load!") // No pages to load! }.collect() } }

No pages to load!

Terminal operators

收集上游流发出的值

collect

可直接调用或接受lambda

suspend fun main() { withContext(Dispatchers.Default) { flowOf(1, 2, 3).collect { println("Collected $it!") } } } suspend fun main() { withContext(Dispatchers.Default) { flowOf(1, 2, 3).onEach { println("Collected $it!") }.collect() } }

collectLatest

收集一个流，但在发出新值时取消未完成的工作

suspend fun main() { withContext(Dispatchers.Default) { flow { println("Emitting Page 1") emit("Page 1") delay(50.milliseconds) println("Emitting Page 2 in quick succession") emit("Page 2") delay(200.milliseconds) println("Emitting Page 3") emit("Page 3") }.flowOn(Dispatchers.IO).collectLatest { println("Starting to process $it!") try { delay(100.milliseconds) } catch (e: CancellationException) { println("Canceled processing $it.") throw e } println("Done processing!") } } }

Emitting Page 1 Starting to process Page 1! Emitting Page 2 in quick succession Canceled processing Page 1. Starting to process Page 2! Done processing! Emitting Page 3 Starting to process Page 3! Done processing!

first

返回第一个发出的值，然后取消收集

suspend fun main() { withContext(Dispatchers.Default) { val firstValue = flowOf(1, 2, 3).first() println(firstValue) // 1 } }

1

toList、toSet

将发出的值收集到一个集合中

suspend fun main() { withContext(Dispatchers.Default) { val list = flowOf(1, 2, 3).toList() println(list) // [1, 2, 3] val set = flowOf(1, 2, 2, 3).toSet() println(set) // [1, 2, 3] } }

[1, 2, 3] [1, 2, 3]

reduce、fold

将发出的值组合成一个结果，fold使用您提供的值作为其起始值，reduce使用第一个发出的值

suspend fun main() { withContext(Dispatchers.Default) { // Uses the first emitted value as the starting value val reduced = flowOf(1, 2, 3).reduce { accumulator, value -> accumulator + value } // Starts with the provided starting value val folded = flowOf(1, 2, 3).fold(2) { accumulator, value -> accumulator + value } println(reduced) // 6 println(folded) // 8 } }

6 8

launchIn

在特定的协程作用域中收集流，如下屏幕从StateFlow收集值，并在屏幕关闭时停止收集协程

data class Coordinate(val x: Int, val y: Int) class MyScreen(val scope: CoroutineScope) { private val _mousePosition = MutableStateFlow<Coordinate>(Coordinate(0, 0)) val mousePosition get() = _mousePosition.asStateFlow() init { // Starts collecting the StateFlow in the screen's CoroutineScope mousePosition.onEach { updateStatusBar() }.launchIn(scope) } fun moveMouse(newCoordinate: Coordinate) { _mousePosition.value = newCoordinate } private fun updateStatusBar() { println("Mouse is at ${_mousePosition.value}") } } suspend fun main() { withContext(Dispatchers.Default) { val childScope = CoroutineScope( currentCoroutineContext() + Job(currentCoroutineContext()[Job]) ) val screen = MyScreen(childScope) delay(100.milliseconds) screen.moveMouse(Coordinate(10, 15)) delay(100.milliseconds) screen.moveMouse(Coordinate(1, 3)) delay(100.milliseconds) childScope.cancel() } }

Mouse is at Coordinate(x=0, y=0) Mouse is at Coordinate(x=10, y=15) Mouse is at Coordinate(x=1, y=3)

• 上面launchIn(scope)等价如下代码，在 scope 这个作用域里启动一个协程持续收集 mousePosition，每来一个值就调 updateStatusBar()
• 但这里在 init {} 构造里，用collect 会挂起当前协程，卡住构造，所以用 launchIn 把收集丢到后台协程

scope.launch { mousePosition.onEach { updateStatusBar() }.collect() }