Kotlin Academy · 课时

flowOn 与 buffer：上下文和背压

Q: 「flowOn 与 buffer：上下文和背压」这节课中我会学到什么？

使用 flowOn 更改发送上下文，并使用 buffer 缓冲发送值以应对背压。 你通过在浏览器中直接运行的动手代码来练习 Kotlin Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

使用 flowOn 更改发送上下文，并使用 buffer 缓冲发送值以应对背压。

第 4 / 4 课13 个步骤

flowOn 与 buffer：上下文和背压 是 CoddyKit 上的免费 Kotlin Academy 课时。这是第 4 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 Kotlin Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Kotlin Academy 课程共包含 4 节课。

数据流上下文

默认情况下，数据流会在调用 collect 的协程上下文中运行。除非进行更改，否则生产者和消费者使用相同的调度器。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        println("emit on: ${Thread.currentThread().name}")
        emit(1)
    }.collect {
        println("collect on: ${Thread.currentThread().name}")
    }
}

flowOn 更改上游上下文

flowOn(dispatcher) 会让上游数据流（链中位于它上方的所有部分）在指定的调度器上运行，而收集操作仍在调用方的调度器上进行。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        println("emit: ${Thread.currentThread().name}")
        emit(1)
    }.map {
        println("map: ${Thread.currentThread().name}")
        it * 2
    }.flowOn(Dispatchers.Default)  // above runs on Default
    .collect {
        println("collect: ${Thread.currentThread().name}")
    }
}

链中使用多个 flowOn

您可以多次使用 flowOn。每个 flowOn 都会影响其正上方直到上一个 flowOn 之间的运算符。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow { emit(1) }
        .map { it + 1 }.flowOn(Dispatchers.IO)      // map runs on IO
        .map { it * 2 }.flowOn(Dispatchers.Default)  // this map runs on Default
        .collect { println("Result: $it") }           // collect on Main (runBlocking)
}

背压问题

当生产者发出值的速度快于收集器处理的速度时，值就会排队等待。如果没有缓冲，生产者将被迫等待。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val time = System.currentTimeMillis()
    flow {
        repeat(3) { i ->
            delay(100)  // fast producer
            emit(i)
        }
    }.collect {
        delay(300)      // slow consumer
        println("Got $it in ${System.currentTimeMillis() - time}ms")
    }
}

buffer() 运算符

buffer() 会让生产者和消费者在不同的协程中并发运行，并将发出的值缓存在通道中。生产者无需等待消费者。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val time = System.currentTimeMillis()
    flow {
        repeat(3) { i -> delay(100); emit(i) }
    }.buffer()  // producer and consumer run concurrently
    .collect {
        delay(300)
        println("Got $it in ${System.currentTimeMillis() - time}ms")
    }
}

buffer 容量

buffer(capacity) 会设置通道缓冲区的大小。当缓冲区已满时，生产者会挂起（产生背压）。默认容量为 64。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.channels.Channel
fun main() = runBlocking {
    flow { repeat(5) { emit(it) } }
        .buffer(Channel.RENDEZVOUS)    // 0: producer waits
        // .buffer(Channel.BUFFERED)   // default: 64
        // .buffer(Channel.UNLIMITED)  // unbounded
        .collect { delay(50); println(it) }
}

conflate()：仅保留最新值

当收集器处理速度较慢时，conflate() 会丢弃中间值，只保留最近一次发出的值。这适用于用户界面状态。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        repeat(5) { i -> emit(i); delay(50) }
    }.conflate()
    .collect { i ->
        delay(150)
        println("Collected: $i")  // skips some values
    }
}

collectLatest：处理缓慢的收集器

当新值到达时，collectLatest 会取消当前的收集代码块，并使用最新值重新开始。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        emit(1); delay(50)
        emit(2); delay(50)
        emit(3)
    }.collectLatest { value ->
        println("Processing $value")
        delay(100)  // gets cancelled if new value arrives
        println("Done $value")  // only prints for last value
    }
}

flowOn + buffer 模式

结合使用 flowOn 和 buffer：让生产者在 IO（网络/磁盘）上运行，缓冲结果，然后在主线程上收集。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun fetchItems(): Flow<String> = flow {
    repeat(3) { i ->
        delay(100)  // simulate IO
        emit("item-$i")
    }
}.flowOn(Dispatchers.IO).buffer(10)
fun main() = runBlocking {
    fetchItems().collect { println("UI: $it") }
}

channelFlow：并发生产者

channelFlow 会创建一个由通道支持的数据流，允许构建器内部的多个协程并发发出值。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun concurrentFlow(): Flow<Int> = channelFlow {
    launch { send(1) }
    launch { send(2) }
    launch { send(3) }
}
fun main() = runBlocking {
    concurrentFlow().collect { println(it) }
}

选择合适的策略

总结：使用 flowOn 切换上下文，使用 buffer 提高吞吐量，使用 conflate 为用户界面仅保留最新值，使用 collectLatest 取消过时的处理。

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
// Guidelines:
// CPU-heavy production  -> flowOn(Dispatchers.Default)
// IO-heavy production   -> flowOn(Dispatchers.IO)
// Slow consumer         -> buffer()
// UI state updates      -> conflate() or StateFlow
// Search/autocomplete   -> collectLatest or flatMapLatest
fun main() = runBlocking { println("Choose the right strategy!") }

快速检查

flowOn 会影响数据流链中的哪一部分？

在交互式课程中回答

回顾

flowOn 会将上游工作转移到另一个调度器。buffer 将生产者与消费者解耦，以提高吞吐量。conflate 会丢弃中间值。collectLatest 会在新值到达时取消处理缓慢的任务。

免费开始

用 AI 导师学习 Kotlin — 免费

在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。

课程: 51
课程: 203

免费开始学习查看完整课程

常见问题解答

「flowOn 与 buffer：上下文和背压」课时是免费的吗？

是的 — 「flowOn 与 buffer：上下文和背压」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Kotlin Academy 课程的其余内容，请升级到 CoddyKit PRO。 Kotlin Academy 课程共包含 4 节课。

「flowOn 与 buffer：上下文和背压」这节课中我会学到什么？

使用 flowOn 更改发送上下文，并使用 buffer 缓冲发送值以应对背压。你通过在浏览器中直接运行的动手代码来练习 Kotlin Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 Kotlin Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 Kotlin Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 4 节课，共 4 节。

「flowOn 与 buffer：上下文和背压」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 Kotlin Academy 课中编写并运行代码吗？

能。每节 Kotlin Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 Kotlin Academy