Kotlin中级——Flow

FlowsFlows 代表一串随着异步产生的值Emitter产生值中间操作从流中消耗值对其操作并返回另一个流Collector从流中消耗值suspend fun main() { // The emitter produces values flowOf(0x4B, 0x6F, 0x74, 0x6C, 0x69, 0x6E) // The intermediate operator consumes values, // applies an operation, and returns another flow .map { value - value.toChar() } // The collector consumes the transformed values .collect { updatedValue - println(Say $updatedValue!) } }Cold Flows懒加载在collect时开始产生值。每个collector都会触发一个新的、独立的流执行创建flow()返回一个flowTemit()发出值但只有当collect时它才开始产生值flowOf()根据提供的值创建流.asFlow()将iterable转换为流fun main() { // Creates a flow val pageFlow flow { for (page in 1..3) { println(Loading page $page...) // Emits each page as it is loaded emit(Page $page) } } println(Creating a cold flow doesnt run it!) // Creates a flow from provided values val predefinedPageFlow flowOf(Page 1, Page 2, Page 3) // Creates a flow from a range val generatedPageFlow (1..3).asFlow() }Creating a cold flow doesnt run it!collectsuspend fun main() { withContext(Dispatchers.Default) { val pageFlow flow { for (page in 1..3) { println(Loading page $page...) emit(Page $page) } } // Collects the flow with a lambda that receives each emitted page pageFlow.collect { page - println(Processing $page...) delay(100.milliseconds) println(Done processing $page.) } } }Loading page 1... Processing Page 1... Done processing Page 1. Loading page 2... Processing Page 2... Done processing Page 2. Loading page 3... Processing Page 3... Done processing Page 3.中间操作可自定义用于flow的中间操作// A simplified custom implementation of the default .map() operator fun T, R FlowT.myMap(transform: suspend (value: T) - R): FlowR flow { // Collects values from the upstream flow thismyMap.collect { value - // Transforms each collected value and emits the result emit(transform(value)) } } suspend fun main() { // Creates a flow, applies the custom map operator, and collects the transformed values flowOf(1, 2, 3).myMap { 2 * it }.collect { println(Collecting $it) } }Collecting 2 Collecting 4 Collecting 6调用suspend方法flow继承上下文协程不能启动新协程或调用withContext()切换协程上下文来调用emit()suspend fun loadPage(): Int { delay(100) return 3 } suspend fun main() { flow { emit(loadPage()) }.collect { println(it) // 3 } }flowOn()更改上游流的协程上下文保持下游流的协程上下文suspend fun main() { withContext(Dispatchers.Default CoroutineName(downstream)) { flow { val coroutineName currentCoroutineContext()[CoroutineName]?.name // Emits in the coroutine context applied with .flowOn() println(Emitting 1 in $coroutineName) // Emitting 1 in upstream emit(1) // Changes the coroutine context of the upstream flow }.flowOn(Dispatchers.IO CoroutineName(upstream)) .collect { val coroutineName currentCoroutineContext()[CoroutineName]?.name // Collects in the callers coroutine context println(Collecting $it in $coroutineName) // Collecting 1 in downstream } } }如上flow运行在upstream协程collect仍运行在downstream协程Emitting 1 in upstream Collecting 1 in downstream处理异常emit和collect都可能抛出异常如果在流中不处理异常则异常会向上游传播并被抛出给collect()的调用者class CoroutinesBasicsTest { class MyFlowException(message: String) : Exception(message) suspend fun main() { val myFlow flow { try { // The emit() function calls the lambda passed to collect() emit(a) } catch (e: MyFlowException) { println(Collector threw $e) // Rethrows the downstream exception throw e } } // Wraps flow collection in try-catch try { myFlow.collect { // Throws an exception from the collect() lambda throw MyFlowException(Cant process $it!) } } catch (e: MyFlowException) { println(Flow collection failed with $e) // Rethrows the exception to the caller throw e } } Test fun runCoroutinesExample() runBlocking { println(——————————————————————————————————————————————————) main() println(——————————————————————————————————————————————————) } }如上emit()调用了 collect() 的 lambdalambda 抛了异常emit() 这次调用也就跟着抛出这个异常Collector threw com.example.demo1.coroutines.CoroutinesBasicsTest$MyFlowException: Cant process a! Flow collection failed with com.example.demo1.coroutines.CoroutinesBasicsTest$MyFlowException: Cant process a! Cant process a! com.example.demo1.coroutines.CoroutinesBasicsTest$MyFlowException: Cant process a!可以使用.catch()运算符来处理上游流的异常但不处理collect()抛出的异常suspend fun main() { flow { emit(a) emit(b) // Throws an exception from the upstream flow throw UnsupportedOperationException( I am tired of listing letters ) }.catch { upstreamException - println(Upstream completed with $upstreamException!) // Emits a fallback value downstream emit(Upstream terminated with an exception!) }.collect { println(Got $it) } }Got a Got b Upstream completed with java.lang.UnsupportedOperationException: I am tired of listing letters! Got Upstream terminated with an exception!retry()retry()返回true时重新启动collect最多达到指定的重试次数返回false将停止重试并重新抛出异常sealed interface LoadingState { sealed interface Terminal: LoadingState object Started: LoadingState data class Percentage(val percents: Int): LoadingState object Failed: Terminal object Done: Terminal } fun loadBlob(url: String) flow { emit(LoadingState.Started) val failureChancePerStep 1 - java.lang.Math.pow(0.99, 10.0) repeat(10) { step - if (Random.nextDouble() failureChancePerStep) throw IOException(Failed to load!) emit(LoadingState.Percentage((step 1) * 10)) delay(10.milliseconds) } emit(LoadingState.Done) }.retry(3) { e - if (e is IOException) { // This is an expected error // Waits for one second before retrying delay(1.seconds) true } else { // Stops retrying and rethrows unexpected exceptions false } } suspend fun main() { loadBlob(https://example.org/).collect { println(Got $it) } }Got LoadingState$Started45ff54e6 Got Percentage(percents10) Got Percentage(percents20) Got Percentage(percents30) Got Percentage(percents40) Got Percentage(percents50) Got Percentage(percents60) Got Percentage(percents70) Got Percentage(percents80) Got Percentage(percents90) Got LoadingState$Started45ff54e6 Got Percentage(percents10) Got Percentage(percents20) Got Percentage(percents30) Got LoadingState$Started45ff54e6 Got Percentage(percents10) Got Percentage(percents20) Got Percentage(percents30) Got Percentage(percents40) Got Percentage(percents50) Got Percentage(percents60) Got Percentage(percents70) Got Percentage(percents80) Got Percentage(percents90) Got Percentage(percents100) Got LoadingState$Done238d8e77取消collect()和协程相关联当该协程被取消时flows也被取消val myFlow flow { var i 0 try { while (true) { println(Emitting $i) emit(i) println(Emitted $i) i delay(10.milliseconds) } } catch (e: Throwable) { println(Upstream finished with $e) throw e } } suspend fun main() { coroutineScope { val job launch { try { myFlow.collect { println(Processing $it) delay(5.milliseconds) } } catch (e: Throwable) { println(Collection finished with $e) throw e } } delay(100.milliseconds) // Cancels the coroutine that collects the flow job.cancel() } }Emitting 0 Processing 0 Emitted 0 Emitting 1 Processing 1 Emitted 1 Emitting 2 Processing 2 Emitted 2 Emitting 3 Processing 3 Emitted 3 Emitting 4 Processing 4 Emitted 4 Emitting 5 Processing 5 Upstream finished with kotlinx.coroutines.JobCancellationException: StandaloneCoroutine was cancelled; jobcoroutine#1:StandaloneCoroutine{Cancelling}feed967 Collection finished with kotlinx.coroutines.JobCancellationException: StandaloneCoroutine was cancelled; jobcoroutine#1:StandaloneCoroutine{Cancelling}feed967cancel()会连带取消整个协程及其子协程从 collect 里 throw CancellationException()可以只取消上游流所在的协程继续运行如take()在获取固定数量的值后停止收集fun T FlowT.myTake(count: Int): FlowT flow { require(count 0) val cancellationException CancellationException() var elementsRemaining count try { thismyTake.collect { emit(it) --elementsRemaining if (elementsRemaining 0) { // Cancels the upstream flow after the requested number of values throw cancellationException } } } catch (e: Throwable) { if (e cancellationException) { // Handles the CancellationException used to cancel the upstream flow // Completes the flow after the set number of values in .myTake() } else { // Rethrows unexpected exceptions throw e } } } suspend fun main() { (0..1000).asFlow().myTake(3).collect { println(Got $it) } }Got 0 Got 1 Got 2channelFlow()flow只能从它自己所在的协程里发值不能开子协程去 emit()channelFlow()将多个协程的值发送到同一个流中内部使用send()如下开两个协程分别调用flow1和flow2的collectfun T FlowT.myMerge(other: FlowT): FlowT channelFlow { // CoroutineScope and SendChannel are available as receivers here // Launches a coroutine that collects the receiver flow launch { // Collects the receiver flow thismyMerge.collect { send(it) } } launch { // Launches a coroutine that collects the other flow other.collect { // Calls SendChannel.send send(it) } } } suspend fun main() { val flow1 (0..3).asFlow().onEach { delay(20.milliseconds) } val flow2 (6..9).asFlow().onEach { delay(50.milliseconds) } flow1.myMerge(flow2).collect { println(it) } }0 1 6 2 3 7 8 9channelFlow()自带缓冲区在collect之前发送值默认为64个缓冲区已满时会暂停发送直到缓冲区有空间suspend fun main() { val oneHundredNumbers channelFlow { repeat(100) { println(Sending $it) send(it) } } // Uses the default buffer capacity oneHundredNumbers.collect { println(Processing $it) delay(10.milliseconds) } // Removes the buffer so sending and processing interleave from the start oneHundredNumbers.buffer(0).collect { println(Processing $it) delay(10.milliseconds) } }打印类似如下一个会先发64个再处理一个会边发边处理Sending 0 Sending 1 ...... Sending 65 Processing 0 Sending 66 Processing 1 Sending 67 ...... Sending 0 Processing 0 Sending 1 Processing 1Hot Flows独立于collector发出值的共享流SharedFlow向多个订阅者广播值当您需要广播随时间发生的事件时例如消息或通知请使用它使用MutableSharedFlow创建SharedFlow使用.asSharedFlow()函数公开只读的SharedFlowreplay参数可设置新订阅者接收的先前发射次数热流没有关闭或取消操作。取消只会阻止相应的订阅者collect。要停止新的emit请取消为热流产生值的协程或作用域data class Message( val senderId: Int, val time: Instant, val text: String, ) // Sets the number of already emitted messages that new subscribers receive on subscription const val MESSAGES_TO_REMEMBER 10 class Chatroom { // Stores the SharedFlow in a private backing property private val _messages MutableSharedFlowMessage( // Replays the set amount of last emitted messages to new subscribers replay MESSAGES_TO_REMEMBER ) // Exposes a read-only SharedFlow to subscribers val messages: SharedFlowMessage get() _messages.asSharedFlow() // Emits the message to subscribers suspend fun sendMessageToEveryone(message: Message) { _messages.emit(message) } } suspend fun main() { val nUsers 3 val chatroom Chatroom() withContext(Dispatchers.Default) { // Starts a message reader for each user val messageReaders List(nUsers) { userId - // Starts collection before messages are emitted launch(start CoroutineStart.UNDISPATCHED) { chatroom.messages.collect { message - println(User $userId received $message) } } } // Sends a greeting from each user repeat(nUsers) { userId - chatroom.sendMessageToEveryone( Message( userId, Clock.System.now(), Hello from $userId! ) ) } // Delays to make sure people have enough time to chat delay(100.milliseconds) // Cancels readers because SharedFlow collection doesnt finish by itself messageReaders.forEach { it.cancel() } } }UNDISPATCHED 让这个协程立刻执行到第一个挂起点即到collect父协程 repeat 三次 emit发完就结束了三个collect的子协程会一直挂起。withContext 必须等这些子协程都结束才能返回所以需要调用 cancelUser 0 received Message(senderId0, time2026-08-21T11:28:46.752326697Z, textHello from 0!) User 0 received Message(senderId1, time2026-08-21T11:28:46.755476174Z, textHello from 1!) User 0 received Message(senderId2, time2026-08-21T11:28:46.755488042Z, textHello from 2!) User 1 received Message(senderId0, time2026-08-21T11:28:46.752326697Z, textHello from 0!) User 1 received Message(senderId1, time2026-08-21T11:28:46.755476174Z, textHello from 1!) User 1 received Message(senderId2, time2026-08-21T11:28:46.755488042Z, textHello from 2!) User 2 received Message(senderId0, time2026-08-21T11:28:46.752326697Z, textHello from 0!) User 2 received Message(senderId1, time2026-08-21T11:28:46.755476174Z, textHello from 1!) User 2 received Message(senderId2, time2026-08-21T11:28:46.755488042Z, textHello from 2!)StateFlow存储一个状态值并在该值被新值替换时发出更新。新订阅者一开始收集就会收到当前值然后每次状态更新时都会收到新值。可用于表示随时间变化的状态例如加载进度、UI状态或对象的状态使用内部的value属性设置值其是线程安全的相同值不会重复通知别在 StateFlow 里放可变对象如MutableList的add并不会更新sealed interface LoadingState { sealed interface Terminal: LoadingState object Started: LoadingState data class Percentage(val percents: Int): LoadingState object Failed: Terminal object Done: Terminal } fun loadBlob(url: String): StateFlowLoadingState { // Creates a mutable StateFlow with the initial loading state val result MutableStateFlowLoadingState(LoadingState.Started) DownloadManager.startLoading( url, onPercentageLoaded { percentage - // Replaces the current state with the latest progress result.value LoadingState.Percentage(percentage) }, onCompletion { // Replaces the current state with the completion state result.value LoadingState.Done }, onFailure { // Replaces the current state with the failure state result.value LoadingState.Failed } ) // Exposes the loading state as a read-only StateFlow return result.asStateFlow() } // Defines a callback-based API that downloads data asynchronously object DownloadManager { // Starts loading the url asynchronously fun startLoading( url: String, onPercentageLoaded: (Int) - Unit, onCompletion: () - Unit, onFailure: (Throwable) - Unit ) { // Uses GlobalScope for illustrative purposes only, // to keep this example self-contained GlobalScope.launch { val failureChancePerStep 1 - java.lang.Math.pow(0.99, 10.0) repeat(10) { step - if (Random.nextDouble() failureChancePerStep) { onFailure(IOException(Failed to load!)) returnlaunch } onPercentageLoaded((step 1) * 10) delay(10.milliseconds) } onCompletion() } } } suspend fun main() { loadBlob(https://example.com/).onEach { state - when (state) { is LoadingState.Started - { // Waits for progress updates } is LoadingState.Percentage - println(Loaded ${state.percents}...) is LoadingState.Failed - println(Loading failed.) is LoadingState.Done - println(Finished loading!) } }.takeWhile { it !is LoadingState.Terminal }.collect() }Loaded 10... Loaded 20... Loaded 30... Loaded 40... Loaded 50... Loaded 60... Loaded 70... Loaded 80... Loaded 90... Loaded 100... Finished loading!当使用旧值计算新值时应使用update保证原子性class Post(val id: Long) { // Stores the current number of likes as a StateFlow private val _numberOfLikes MutableStateFlowInt( // Sets the initial number of likes 0 ) // Exposes a read-only StateFlow with the current number of likes val numberOfLikes: StateFlowInt get() _numberOfLikes.asStateFlow() // Adds a like fun like() { // Increments the number of likes atomically for concurrent and multithreaded calls _numberOfLikes.update { it 1 } } } suspend fun drawUpdatedNumberOfLikes(likes: Int) { // Displays the latest number of likes println(${Clock.System.now()}: the number of likes is $likes) } suspend fun main() { withContext(Dispatchers.Default) { val post Post(15) val notifyingJob launch { post.numberOfLikes.collect { drawUpdatedNumberOfLikes(it) } } // Simulates users who like the post coroutineScope { repeat(10) { launch { delay(Random.nextInt(100).milliseconds) post.like() } } } // Cancels collection after all simulated users finish notifyingJob.cancelAndJoin() } }2026-08-25T03:35:20.967404757Z: the number of likes is 0 2026-08-25T03:35:20.999675715Z: the number of likes is 1 2026-08-25T03:35:21.006477386Z: the number of likes is 2 2026-08-25T03:35:21.007046583Z: the number of likes is 3 2026-08-25T03:35:21.018720338Z: the number of likes is 4 2026-08-25T03:35:21.020541355Z: the number of likes is 5 2026-08-25T03:35:21.030318383Z: the number of likes is 6 2026-08-25T03:35:21.033299043Z: the number of likes is 7 2026-08-25T03:35:21.037435797Z: the number of likes is 8 2026-08-25T03:35:21.047466551Z: the number of likes is 9 2026-08-25T03:35:21.071606167Z: the number of likes is 10如果需要累积通知需要利用之前的旧数据创建新数据并调用updatedata class Message( val senderId: Int, val time: Instant, val text: String, ) class Chatroom { // Stores the full message history private val _messageHistory MutableStateFlowListMessage(emptyList()) // Exposes a read-only StateFlow with the current message history val messageHistory: StateFlowListMessage get() _messageHistory.asStateFlow() // Sends a message to all subscribers of the messageHistory flow suspend fun sendMessageToEveryone(message: Message) { // Adds the new message to the current history atomically _messageHistory.update { it message } } } suspend fun main() { val nUsers 3 val chatroom Chatroom() withContext(Dispatchers.Default) { // Starts a message reader for each user val messageReaders List(nUsers) { userId - launch(start CoroutineStart.UNDISPATCHED) { chatroom.messageHistory.collect { currentHistory - println(User $userId sees the history as $currentHistory) } } } // Sends a greeting from each user repeat(nUsers) { userId - chatroom.sendMessageToEveryone( Message( userId, Clock.System.now(), Hello from $userId! ) ) } // Delays to make sure users have enough time to receive updates delay(100.milliseconds) // Cancels readers because StateFlow collection doesnt finish by itself messageReaders.forEach { it.cancel() } } }User 0 sees the history as [] User 1 sees the history as [] User 2 sees the history as [] User 0 sees the history as [Message(senderId0, time2026-08-25T03:40:10.636117883Z, textHello from 0!)] User 0 sees the history as [Message(senderId0, time2026-08-25T03:40:10.636117883Z, textHello from 0!), Message(senderId1, time2026-08-25T03:40:10.644839868Z, textHello from 1!), Message(senderId2, time2026-08-25T03:40:10.644885405Z, textHello from 2!)] User 1 sees the history as [Message(senderId0, time2026-08-25T03:40:10.636117883Z, textHello from 0!), Message(senderId1, time2026-08-25T03:40:10.644839868Z, textHello from 1!), Message(senderId2, time2026-08-25T03:40:10.644885405Z, textHello from 2!)] User 2 sees the history as [Message(senderId0, time2026-08-25T03:40:10.636117883Z, textHello from 0!), Message(senderId1, time2026-08-25T03:40:10.644839868Z, textHello from 1!), Message(senderId2, time2026-08-25T03:40:10.644885405Z, textHello from 2!)]当数据量大时每次重新拷贝创建新数据太耗时可以使用PersistentList其add返回一个新的 PersistentList且更高效private val _messages MutableStateFlowPersistentListMessage(persistentListOf()) val messages: StateFlowListMessage _messages.asStateFlow() suspend fun sendMessageToEveryone(message: Message) { _messages.update { it.add(message) } // 返回新实例不改旧的 }将Cold Flows转为SharedFlowshareIn()可控制上游收集何时开始和停止的选项以及新用户接收到的先前排放量需提供协程作用域SharingStarted控制上游收集何时开始和停止replay用于控制新用户接收的先前值data class Message( val senderId: Int, val time: Instant, val text: String, ) class Chatroom { // Stores the message flow private val _messages MutableSharedFlowMessage() // Exposes a read-only SharedFlow with emitted messages // New subscribers dont receive already emitted messages val messages: SharedFlowMessage get() _messages.asSharedFlow() // Sends a message to all subscribers of the messages flow suspend fun sendMessageToEveryone(message: Message) { _messages.emit(message) } } suspend fun main() { val nUsers 3 val chatroom Chatroom() withContext(Dispatchers.Default) { // Creates a child scope of the currently running coroutine val derivedFlowsScope CoroutineScope( currentCoroutineContext() Job(currentCoroutineContext()[Job]) ) // Shares serialized messages between subscribers val serializedMessages: SharedFlowString chatroom .messages .map { // Serializes each message once for the shared flow senderId: ${it.senderId}, time: ${it.time}, text: Base64.Default.encode(it.text.encodeToByteArray()) } .shareIn( // Starts the sharing coroutine in this scope. // The upstream flow, including .map(), runs in that coroutine derivedFlowsScope, // Starts collecting the upstream flow immediately, // before the first subscriber appears SharingStarted.Eagerly, // Doesnt replay previous serialized messages to new subscribers replay 0, ) // Starts a message reader for each user val messageReaders List(nUsers) { userId - launch(start CoroutineStart.UNDISPATCHED) { serializedMessages.collect { serializedMessage - println(User $userId observes the message $serializedMessage) } } } // Sends a greeting from each user repeat(nUsers) { userId - chatroom.sendMessageToEveryone( Message( userId, Clock.System.now(), Hello from $userId! ) ) } // Delays to make sure users have enough time to receive updates delay(100.milliseconds) // Cancels readers because SharedFlow collection doesnt finish by itself messageReaders.forEach { it.cancel() } // Cancels the scope that runs the derived hot flow derivedFlowsScope.cancel() } }CoroutineScope(currentCoroutineContext() Job(currentCoroutineContext()[Job]))相当于创建自身的作用域同时不影响父类代码继续运行但会跟随父类一起cancelUser 0 observes the message senderId: 0, time: 2026-08-25T06:59:19.575722935Z, text: SGVsbG8gZnJvbSAwIQ User 0 observes the message senderId: 1, time: 2026-08-25T06:59:19.578971274Z, text: SGVsbG8gZnJvbSAxIQ User 0 observes the message senderId: 2, time: 2026-08-25T06:59:19.596794468Z, text: SGVsbG8gZnJvbSAyIQ User 1 observes the message senderId: 0, time: 2026-08-25T06:59:19.575722935Z, text: SGVsbG8gZnJvbSAwIQ User 1 observes the message senderId: 1, time: 2026-08-25T06:59:19.578971274Z, text: SGVsbG8gZnJvbSAxIQ User 1 observes the message senderId: 2, time: 2026-08-25T06:59:19.596794468Z, text: SGVsbG8gZnJvbSAyIQ User 2 observes the message senderId: 0, time: 2026-08-25T06:59:19.575722935Z, text: SGVsbG8gZnJvbSAwIQ User 2 observes the message senderId: 1, time: 2026-08-25T06:59:19.578971274Z, text: SGVsbG8gZnJvbSAxIQ User 2 observes the message senderId: 2, time: 2026-08-25T06:59:19.596794468Z, text: SGVsbG8gZnJvbSAyIQ将Cold Flows转为StateFlow.stateIn()类似但需要一个初始值val lastUpdateFlow: StateFlowInstant? chatroom .messageHistory .map { currentHistory - currentHistory.lastOrNull()?.time } .stateIn( // Starts the sharing coroutine in this scope // The upstream flow, including .map(), runs in that coroutine derivedFlowsScope, // Starts collecting when the first subscriber appears // and stops when the last subscriber disappears SharingStarted.WhileSubscribed(), // Sets the initial state before the first upstream emission null, )取消热流当您取消收集热流的协程时您只会取消该订阅者热流本身没有取消操作。要取消热流请取消为其生成值的协程或作用域处理异常MutableSharedFlow、MutableStateFlow的代码抛出异常请在运行该代码的协程中处理它如果订阅者在收集时抛出异常请在收集协程中处理它使用.shareIn()或.stateIn()扩展函数创建的热流若上流抛出异常则会取消协程suspend fun main() { withContext(Dispatchers.Default) { launch { flowInt { error(An upstream failure) }.stateIn( thislaunch ) } } }Exception in thread main java.lang.IllegalStateException: An upstream failure失败后可以重新启动上游收集。将.retry()运算符放在.shareIn()或.stateIn()之前suspend fun main() { coroutineScope { launch { var currentAttempt 0 val stateFlow flow { delay(10.milliseconds) if (currentAttempt 5) { println(An error happened!) error(An upstream failure) } else { println(Success.) emit(10) } } // Restarts the upstream flow after recoverable failures .retry(retries 5) .stateIn( // Starts the sharing coroutine in this scope thislaunch ) stateFlow.collect { println(Observed $it) // Cancels collection and the sharing coroutine thislaunch.cancel() } } } }An error happened! An error happened! An error happened! An error happened! An error happened! Success. Observed 10