第 9 章协程基础（launch / async / Flow）

学习目标：理解协程为什么比线程轻；会用 launch / async / await 写并发代码；理解结构化并发（CoroutineScope）；会用 Flow 处理流式数据；摆脱回调地狱。

9.1 痛点：传统并发的"原罪"

9.1.1 阻塞线程的浪费

kotlin

// 同步阻塞写法
fun loadUser(): String {
    Thread.sleep(2000)         // 模拟 IO 等待 2 秒
    return "Alice"
}

fun main() {
    println(loadUser())        // 等 2 秒
}

问题：这 2 秒里，整个线程都在挂着什么也不做。如果同时有 1000 个用户请求，要 1000 个线程，每个 ~1MB 栈空间 = 1GB 内存，而且都在干等。

9.1.2 异步回调地狱

kotlin

// 异步回调写法（Java 风格）
fetchUser(userId, callback = { user ->
    fetchOrders(user.id, callback = { orders ->
        fetchPayments(orders[0].id, callback = { payments ->
            updateUI(payments)
            // 缩进 4 层，再多就爆炸
        })
    })
})

问题：写起来恶心，错误处理更恶心，调试栈跟踪基本看不懂。

9.2 协程：写同步代码，跑异步效果

9.2.1 第一个协程

kotlin

import kotlinx.coroutines.*

fun main() = runBlocking {
    println("start")
    delay(2000)                 // 挂起 2 秒，但不阻塞线程
    println("end")
}

跟 Thread.sleep 的差别：

Thread.sleep(2000)：线程被挂起，不能干别的
delay(2000)：协程被挂起，线程被释放去做别的协程

9.2.2 launch：启动一个并发任务

kotlin

fun main() = runBlocking {
    println("main start")
    
    launch {
        delay(1000)
        println("child done at ${System.currentTimeMillis()}")
    }
    
    launch {
        delay(500)
        println("another child done at ${System.currentTimeMillis()}")
    }
    
    println("main end (但等子协程完成才退出)")
}

输出：

main start
main end (但等子协程完成才退出)
another child done at ...   ← 500ms
child done at ...           ← 1000ms

9.2.3 async / await：要返回值的并发

kotlin

fun main() = runBlocking {
    val time = measureTimeMillis {
        val a = async { fetchValueA() }     // 同时启动
        val b = async { fetchValueB() }
        println("a + b = ${a.await() + b.await()}")
    }
    println("总耗时: ${time}ms")
}

suspend fun fetchValueA(): Int { delay(1000); return 10 }
suspend fun fetchValueB(): Int { delay(1000); return 20 }

输出：

a + b = 30
总耗时: ~1010ms     ← 不是 2000ms！两个 fetch 是并发的

9.3 suspend 函数：协程的"暂停点"

kotlin

suspend fun fetchUser(): User {
    delay(1000)
    return User("Alice")
}

suspend 修饰符告诉编译器："这个函数可能挂起，只能在协程或另一个 suspend 函数里调用"。

9.3.1 编译器的魔法：CPS 变换

kotlin

// 你写的代码
suspend fun loadUser(): User {
    val data = fetchData()
    return parseUser(data)
}

编译器编译成"状态机"：

kotlin

// 简化的伪代码
fun loadUser(continuation: Continuation<User>): Any {
    when (state) {
        0 -> {
            state = 1
            return fetchData(continuation)    // 挂起，返回 SUSPEND
        }
        1 -> {
            return parseUser(result)
        }
    }
}

📌 本质：suspend 函数被编译成不阻塞的状态机，每个挂起点保存上下文，恢复时从那里继续跑。

9.3.2 怎么调 suspend 函数？

只能在以下地方调：

另一个 suspend 函数 里
协程构建器 里：runBlocking { }、launch { }、async { }

kotlin

suspend fun a() { delay(100) }
suspend fun b() { a() }                     // ✅ suspend 调 suspend

fun main() {
    a()                                      // ❌ 编译报错
    runBlocking { a() }                     // ✅ 在协程里调
}

9.4 协程作用域 CoroutineScope

9.4.1 为啥需要 Scope

每个协程都需要一个"管理者"，告诉它"在哪个线程跑、生命周期挂在哪、出错怎么传播"。这就是 CoroutineScope。

kotlin

val scope = CoroutineScope(Dispatchers.IO)

scope.launch { /* 在 IO 线程池跑 */ }
scope.launch { /* ... */ }

// 一键取消所有子协程
scope.cancel()

9.4.2 常见 Dispatcher

Dispatcher	用途
`Dispatchers.Main`	UI 线程（Android / Swing）
`Dispatchers.IO`	文件 / 网络 IO
`Dispatchers.Default`	CPU 密集任务（JSON 解析、计算）
`Dispatchers.Unconfined`	不限定线程，从哪来回哪去

kotlin

launch(Dispatchers.IO) {
    val data = readFromDisk()       // 在 IO 线程读
    withContext(Dispatchers.Main) {
        updateUI(data)              // 切回主线程更新 UI
    }
}

9.4.3 结构化并发（Structured Concurrency）

Kotlin 协程的核心概念：子协程的生命周期受父协程管理。

kotlin

suspend fun fetchAll() = coroutineScope {       // 创建子作用域
    val a = async { fetchA() }
    val b = async { fetchB() }
    a.await() + b.await()
    // 这里如果 fetchA 抛异常，fetchB 自动被取消
    // 整个 coroutineScope 等所有子协程结束才返回
}

📌 跟 Java 线程的本质区别：
Java：起了个线程，忘记 join 它就成野线程
Kotlin：协程必须挂在某个 scope 上，scope 取消时所有子协程一起取消，没有泄漏

9.5 错误处理

9.5.1 try-catch（同步代码风格）

kotlin

suspend fun safeLoadUser(): User? = try {
    fetchUser()
} catch (e: Exception) {
    println("加载失败: ${e.message}")
    null
}

9.5.2 SupervisorJob：子协程互不影响

kotlin

val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)

scope.launch { throw RuntimeException("A 挂了") }     // A 挂不影响 B
scope.launch { delay(1000); println("B 跑完了") }

9.5.3 CoroutineExceptionHandler

kotlin

val handler = CoroutineExceptionHandler { _, e ->
    println("捕获到异常: ${e.message}")
}

CoroutineScope(Dispatchers.IO + handler).launch {
    throw RuntimeException("oops")
}

9.6 Flow：异步数据流

9.6.1 痛点：单次值 vs 多个值

suspend + async 处理的是"一次性返回 1 个值"。但有些场景是"陆续返回多个值"：

WebSocket 消息流
数据库查询结果（分页）
UI 状态变更
文件按行读

9.6.2 Flow 入门

kotlin

import kotlinx.coroutines.flow.*

fun numbers(): Flow<Int> = flow {
    for (i in 1..5) {
        delay(500)               // 模拟生产
        emit(i)                   // 发出一个值
    }
}

fun main() = runBlocking {
    numbers().collect { value ->
        println("收到: $value")
    }
}
// 每 500ms 收到一个，共 5 个

9.6.3 Flow 的"冷"特性

Flow 是"冷的"：只有 collect 调用时才开始执行。多次 collect 会重新执行 flow 块。

kotlin

val f = flow {
    println("flow 开始")
    emit(1); emit(2); emit(3)
}

f.collect { println("第 1 次: $it") }   // 触发 "flow 开始" + 1, 2, 3
f.collect { println("第 2 次: $it") }   // 又触发一遍

9.6.4 Flow 操作符（跟集合很像）

kotlin

flow { repeat(10) { emit(it) } }
    .filter { it % 2 == 0 }
    .map { it * it }
    .take(3)
    .collect { println(it) }
// 0, 4, 16

9.6.5 StateFlow 和 SharedFlow

类型	用途
`Flow<T>`	冷流，每次 collect 重新跑
`StateFlow<T>`	热流，保留当前值，新订阅者立即收到。适合 UI 状态
`SharedFlow<T>`	热流，可配置缓冲。适合事件广播

kotlin

class UiViewModel {
    private val _state = MutableStateFlow(UiState.Loading)
    val state: StateFlow<UiState> = _state
    
    suspend fun load() {
        _state.value = UiState.Loading
        val data = fetchData()
        _state.value = UiState.Success(data)
    }
}

9.7 实战：并发请求 + 超时

kotlin

import kotlinx.coroutines.*

suspend fun fetchProduct(id: String): String {
    delay(500)
    return "Product[$id]"
}

suspend fun fetchAllConcurrently(ids: List<String>): List<String> = coroutineScope {
    ids.map { id ->
        async { fetchProduct(id) }
    }.awaitAll()
}

fun main() = runBlocking {
    val time = System.currentTimeMillis()
    val products = withTimeout(2000) {
        fetchAllConcurrently(listOf("p1", "p2", "p3", "p4", "p5"))
    }
    println("拿到 ${products.size} 个商品，耗时 ${System.currentTimeMillis() - time}ms")
}

输出：

拿到 5 个商品，耗时 ~510ms        ← 5 个并发，不是 5*500=2500ms

9.8 章末小结

              ★ 第 9 章核心知识图谱 ★
                       │
        ┌──────────────┼──────────────┐
        │              │              │
    ┌──▼──┐        ┌──▼──┐       ┌──▼──┐
    │ 启动 │        │ 同步 │       │ 流  │
    ├─────┤        ├─────┤       ├─────┤
    │launch│        │suspend│     │Flow  │
    │async │        │delay  │     │collect│
    │await │        │with   │     │State │
    │coScop│        │Context│     │Shared│
    └─────┘        └─────┘       └─────┘
                       │
                  ┌────▼────┐
                  │ 结构化  │
                  │ 并发     │
                  ├─────────┤
                  │ Scope   │
                  │ Cancel  │
                  │ 异常传播 │
                  └─────────┘

🎤 9.9 章末面试题（10 道）

Q1. 协程跟线程有啥区别？

答：

维度	线程（Thread）	协程（Coroutine）
本质	OS 内核调度	用户态调度
创建成本	~1MB / 个	几 KB / 个
阻塞代价	线程被占用	让出线程，复用
数量级	千级	百万级
切换	内核态	用户态
错误处理	try-catch + 回调	try-catch（同步风格）

简单说：协程是"在线程上跑的轻量任务"，N 个协程可以共享 M 个线程（N >> M）。

Q2. `suspend` 关键字背后做了什么？

答：编译器把 suspend 函数编译成状态机（CPS 变换 - Continuation-Passing Style）：

函数实际签名加了一个 Continuation 参数
函数体被切成多个状态，每个挂起点保存当前状态
挂起时返回 COROUTINE_SUSPENDED，不阻塞线程
异步操作完成后调用 continuation.resume(result) 恢复执行

所以 suspend 函数本质是异步代码的同步写法，不是 Java Thread.sleep 那种阻塞。

Q3. `launch` 和 `async` 啥区别？

答：

维度	launch	async
返回值	`Job`（无结果）	`Deferred<T>`（带结果）
取结果	不能	`.await()` 阻塞等待
异常处理	立即抛到 scope	`.await()` 时才抛
典型用途	"fire and forget"	并发计算后合并结果

kotlin

launch { doSomething() }                          // 起后不管
val r = async { compute() }.await()              // 起后等结果

Q4. 什么是"结构化并发"？

答：协程必须挂在某个 CoroutineScope 上，scope 的生命周期决定子协程的生命周期：

Scope 取消 → 所有子协程取消
父协程等所有子协程结束才结束
一个子协程异常 → 默认情况下取消所有兄弟协程
没有"野协程"

这跟 Java Thread 的"启动后忘记"形成鲜明对比。所有泄漏的可能性都被堵死。

Q5. `withContext` 用来做什么？

答：切换 Dispatcher。常见场景：

kotlin

suspend fun loadData() {
    val data = withContext(Dispatchers.IO) {
        readFromDisk()              // 在 IO 线程
    }
    withContext(Dispatchers.Main) {
        updateUI(data)              // 切回主线程
    }
}

withContext 本身是 suspend 函数，等 lambda 执行完才返回。

Q6. `coroutineScope` 跟 `supervisorScope` 区别？

答：

coroutineScope：一个子协程异常 → 取消所有兄弟协程 + 异常往上抛
supervisorScope：一个子协程异常 → 不影响兄弟协程，异常被隔离

kotlin

coroutineScope {
    launch { throw RuntimeException("A") }
    launch { delay(1000); println("B") }   // ❌ 不会执行（被取消）
}

supervisorScope {
    launch { throw RuntimeException("A") }
    launch { delay(1000); println("B") }   // ✅ 正常执行
}

UI 场景常用 supervisorScope（一个组件挂了不要把整个页面拖垮）。

Q7. Flow 是什么？跟 Sequence 啥区别？

答：

维度	Sequence	Flow
是否支持 suspend	❌	✅
是否冷的	✅	✅
调度器	当前线程	可切换
典型场景	同步数据流	异步数据流（网络 / DB / WebSocket）

简单说：Flow = Sequence + suspend，所以能在中间做异步操作。

Q8. `StateFlow` / `SharedFlow` / `Flow` 区别？

答：

类型	冷热	当前值	重播给新订阅者	典型场景
`Flow`	冷	无	重新执行整个 flow	一次性数据流（DB 查询）
`StateFlow`	热	有当前值	当前值立即给	UI 状态（loading/success/error）
`SharedFlow`	热	无（除非配置）	可配置缓冲 N 个	事件广播（toast、navigate）

Q9. 怎么取消一个协程？取消时 suspend 函数怎么响应？

答：

kotlin

val job = launch { 
    repeat(1000) { i ->
        delay(100)              // delay 是协作式取消点
        println(i)
    }
}
delay(500)
job.cancel()                    // 取消
job.join()                      // 等取消完成

协程取消是"协作式"的：

delay()、yield()、withContext() 等 suspend 函数会检查取消状态，被取消时抛 CancellationException
CPU 密集循环不会自动取消，需要手动检查 isActive 或调 yield()

Q10. 协程能解决回调地狱，但 Java 21 的虚拟线程也能，区别在哪？

答：本质思路一致 —— "用户态调度的轻量任务"。差别：

维度	Kotlin 协程	Java 21 Virtual Thread
历史	2018+	2023+
跨平台	JVM / JS / Native	JVM only
语法	suspend 关键字	透明（普通方法签名不变）
类型系统	编译期检查可挂起性	运行时
流式 API	Flow（成熟）	Stream（不变）
取消	结构化并发	InterruptedException
学习成本	中等（要学 Scope/Job/Dispatcher）	低

💡 趋势：未来可能融合 —— Kotlin 协程在 JVM 后端用虚拟线程做底层 Dispatcher。

下一章 → 第 10 章 · 综合实战项目 →

🎬 可视化演示

演示加载缓慢或样式异常？点此在新标签页打开 ↗

💻 示例代码

CoroutineDemo.kt

kotlin

/**
 * 第 9 章 · 协程基础 — 综合示例
 *
 * 涵盖：runBlocking / launch / async / await /
 *      suspend / delay / coroutineScope / supervisorScope /
 *      withContext / Dispatchers / Flow / withTimeout
 *
 * 依赖：implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.1")
 */

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.measureTimeMillis

suspend fun fetchUser(id: String): String {
    delay(500)
    return "User[$id]"
}

suspend fun fetchOrders(userId: String): List<String> {
    delay(700)
    return listOf("Order-1", "Order-2", "Order-3")
}

suspend fun fetchPayment(orderId: String): String {
    delay(300)
    return "Payment[$orderId]=¥100"
}

fun main() = runBlocking {
    println("=== 1. 串行 vs 并发 ===")
    val seqTime = measureTimeMillis {
        val u = fetchUser("u1")
        val o = fetchOrders("u1")
        val p = fetchPayment(o[0])
        println("  串行结果: $u, ${o.size} orders, $p")
    }
    println("  串行耗时: ${seqTime}ms\n")

    val parTime = measureTimeMillis {
        // 这里只能用 async 因为后面要用结果。但 user 和 orders 实际不依赖（演示并发）
        val uDeferred = async { fetchUser("u1") }
        val oDeferred = async { fetchOrders("u1") }
        val u = uDeferred.await()
        val o = oDeferred.await()
        val p = fetchPayment(o[0])
        println("  并发结果: $u, ${o.size} orders, $p")
    }
    println("  并发耗时: ${parTime}ms (节省了 ~500ms)")

    println("\n=== 2. launch：fire and forget ===")
    val job = launch {
        delay(300)
        println("  [子协程] 跑完了")
    }
    println("  [main] 主协程继续往下跑")
    job.join()        // 等子协程结束
    println("  [main] 子协程结束后我才退出")

    println("\n=== 3. async + await：要返回值 ===")
    val total = withContext(Dispatchers.Default) {
        val a = async { delay(200); 10 }
        val b = async { delay(200); 20 }
        a.await() + b.await()
    }
    println("  total = $total (并发执行只用了 ~200ms)")

    println("\n=== 4. 异常处理 ===")
    try {
        coroutineScope {
            launch { throw RuntimeException("子协程 A 挂了") }
            launch {
                delay(500)
                println("  子协程 B 跑完")    // 不会执行！coroutineScope 会取消所有兄弟
            }
        }
    } catch (e: Exception) {
        println("  ✅ 捕获到: ${e.message}")
    }

    println("\n=== 5. supervisorScope：子协程互相隔离 ===")
    supervisorScope {
        launch {
            try {
                throw RuntimeException("子协程 A 挂了")
            } catch (e: Exception) {
                println("  A 自己处理了异常: ${e.message}")
            }
        }
        launch {
            delay(200)
            println("  ✅ 子协程 B 不受 A 影响，正常跑完")
        }
    }

    println("\n=== 6. withTimeout：超时控制 ===")
    try {
        withTimeout(800) {
            delay(2000)        // 超时！
            println("  这条不会打印")
        }
    } catch (e: TimeoutCancellationException) {
        println("  ⏰ 超时触发: ${e.message}")
    }

    println("\n=== 7. 取消协程 ===")
    val cancelJob = launch {
        try {
            repeat(10) { i ->
                delay(200)
                println("  在跑第 $i 次")
            }
        } catch (e: CancellationException) {
            println("  ✋ 协程被取消了")
        }
    }
    delay(500)
    cancelJob.cancelAndJoin()
    println("  cancel 完成")

    println("\n=== 8. Flow：异步数据流 ===")
    fun numbers(): Flow<Int> = flow {
        for (i in 1..5) {
            delay(200)
            emit(i)
            println("    [flow] 发出了 $i")
        }
    }

    println("  开始 collect:")
    numbers()
        .filter { it % 2 == 0 }
        .map { it * it }
        .collect { println("  收到: $it") }

    println("\n=== 9. Flow 的"冷"特性 ===")
    val coldFlow = flow {
        println("    [flow 块] 开始执行")
        emit(1); emit(2)
    }
    println("  第一次 collect:")
    coldFlow.collect { println("    收到 $it") }
    println("  第二次 collect:")
    coldFlow.collect { println("    收到 $it") }
    println("  → flow 块被执行了 2 次（每次 collect 都重跑）")

    println("\n=== 10. 实战：并发批量请求 ===")
    val ids = listOf("p1", "p2", "p3", "p4", "p5")
    val products = ids.map { id ->
        async { fetchUser(id) }
    }.awaitAll()
    println("  并发拿到 ${products.size} 个: $products")

    println("\n=== 11. Dispatchers 切换 ===")
    println("  当前线程: ${Thread.currentThread().name}")
    withContext(Dispatchers.IO) {
        println("  IO 线程: ${Thread.currentThread().name}")
    }
    withContext(Dispatchers.Default) {
        println("  Default 线程: ${Thread.currentThread().name}")
    }
    println("  回到原线程: ${Thread.currentThread().name}")
}

CoroutineDemo.kt ↗

第 9 章 协程基础（launch / async / Flow） ​

9.1 痛点：传统并发的"原罪" ​

9.1.1 阻塞线程的浪费 ​

9.1.2 异步回调地狱 ​

9.2 协程：写同步代码，跑异步效果 ​

9.2.1 第一个协程 ​

9.2.2 launch：启动一个并发任务 ​

9.2.3 async / await：要返回值的并发 ​

9.3 suspend 函数：协程的"暂停点" ​

9.3.1 编译器的魔法：CPS 变换 ​

9.3.2 怎么调 suspend 函数？ ​

9.4 协程作用域 CoroutineScope ​

9.4.1 为啥需要 Scope ​

9.4.2 常见 Dispatcher ​

9.4.3 结构化并发（Structured Concurrency） ​

9.5 错误处理 ​

9.5.1 try-catch（同步代码风格） ​

9.5.2 SupervisorJob：子协程互不影响 ​

9.5.3 CoroutineExceptionHandler ​

9.6 Flow：异步数据流 ​

9.6.1 痛点：单次值 vs 多个值 ​

9.6.2 Flow 入门 ​

9.6.3 Flow 的"冷"特性 ​

9.6.4 Flow 操作符（跟集合很像） ​

9.6.5 StateFlow 和 SharedFlow ​

9.7 实战：并发请求 + 超时 ​

9.8 章末小结 ​

🎤 9.9 章末面试题（10 道） ​

Q1. 协程跟线程有啥区别？ ​

Q2. suspend 关键字背后做了什么？ ​

Q3. launch 和 async 啥区别？ ​

Q4. 什么是"结构化并发"？ ​

Q5. withContext 用来做什么？ ​

Q6. coroutineScope 跟 supervisorScope 区别？ ​

Q7. Flow 是什么？跟 Sequence 啥区别？ ​

Q8. StateFlow / SharedFlow / Flow 区别？ ​

Q9. 怎么取消一个协程？取消时 suspend 函数怎么响应？ ​

Q10. 协程能解决回调地狱，但 Java 21 的虚拟线程也能，区别在哪？ ​

🎬 可视化演示 ​

💻 示例代码 ​

第 9 章协程基础（launch / async / Flow）

9.1 痛点：传统并发的"原罪"

9.1.1 阻塞线程的浪费

9.1.2 异步回调地狱

9.2 协程：写同步代码，跑异步效果

9.2.1 第一个协程

9.2.2 launch：启动一个并发任务

9.2.3 async / await：要返回值的并发

9.3 suspend 函数：协程的"暂停点"

9.3.1 编译器的魔法：CPS 变换

9.3.2 怎么调 suspend 函数？

9.4 协程作用域 CoroutineScope

9.4.1 为啥需要 Scope

9.4.2 常见 Dispatcher

9.4.3 结构化并发（Structured Concurrency）

9.5 错误处理

9.5.1 try-catch（同步代码风格）

9.5.2 SupervisorJob：子协程互不影响

9.5.3 CoroutineExceptionHandler

9.6 Flow：异步数据流

9.6.1 痛点：单次值 vs 多个值

9.6.2 Flow 入门

9.6.3 Flow 的"冷"特性

9.6.4 Flow 操作符（跟集合很像）

9.6.5 StateFlow 和 SharedFlow

9.7 实战：并发请求 + 超时

9.8 章末小结

🎤 9.9 章末面试题（10 道）

Q1. 协程跟线程有啥区别？

Q2. `suspend` 关键字背后做了什么？

Q3. `launch` 和 `async` 啥区别？

Q4. 什么是"结构化并发"？

Q5. `withContext` 用来做什么？

Q6. `coroutineScope` 跟 `supervisorScope` 区别？

Q7. Flow 是什么？跟 Sequence 啥区别？

Q8. `StateFlow` / `SharedFlow` / `Flow` 区别？

Q9. 怎么取消一个协程？取消时 suspend 函数怎么响应？

Q10. 协程能解决回调地狱，但 Java 21 的虚拟线程也能，区别在哪？

🎬 可视化演示

💻 示例代码