Kotlin 协程

发表于 更新于 分类于 ，

概述

Kotlin 协程引入于 Kotlin 1.3，是一种基于线程的轻量级并发工具，适合处理异步任务，基于 kotlinx.coroutines 库。Kotlin 协程支持挂起（suspend）机制，可以挂起和恢复，允许以顺序风格编写异步代码，类似于同步代码，但不会阻塞线程，同时避免传统回调地狱。协程的核心优势包括资源高效（不阻塞线程）和结构化并发（确保资源管理）。每个协程必须在一个作用域（CoroutineScope）中启动。

协程语法点包括：

  1. 启动协程（如 launch、async）
  2. 挂起函数
  3. 上下文和调度器（如 Dispatchers）
  4. 取消和超时
  5. 异常处理
  6. 结构化并发（如 coroutineScope）
  7. 通道（Channel）
  8. 流（Flow）
  9. 选择表达式（select）
  10. 测试工具

协程实际应用场景包括网络请求、数据库操作、并发任务和超时处理。

语法总结

语法点 用途 示例场景
launch 启动无返回结果的协程 后台任务，如日志记录
async 启动返回结果的协程 并行计算结果
runBlocking 阻塞线程等待协程完成 测试或主函数入口
withContext 切换上下文运行挂起块 网络请求或 I/O 操作
suspend 函数 定义可挂起的异步函数 网络请求或 I/O 操作
Dispatchers 选择执行线程（如 IO、Main、Default） UI 更新或数据库操作
Job 和 Deferred 管理任务生命周期和结果 等待任务完成或取消
取消和超时 取消协程或设置时间限制 超时网络请求
异常处理 处理协程中的异常 错误日志记录
coroutineScope 结构化并发管理子协程 确保任务顺序执行
supervisorScope 容错结构化并发 子任务失败不影响其他
Channel 协程间通信 生产者-消费者模式
Flow 处理异步数据流 实时数据更新，如传感器数据
select 选择第一个可用结果 多个异步操作竞争
yield 让出执行权 协作式多任务
joinAll 等待多个 Job 完成 批量任务同步
awaitAll 等待多个 Deferred 结果 并发结果收集
测试工具 测试协程行为 单元测试和集成测试

启动协程

launch

启动不返回结果的协程。

1
2
3
4
5
6
7
8
9
import kotlinx.coroutines.*

fun main() = runBlocking {
    launch {
        delay(1000L)
        println("World!")
    }
    println("Hello,")
}

async

启动返回 Deferred<T> 的协程，用于获取结果。

1
2
3
4
5
6
7
8
9
import kotlinx.coroutines.*

fun main() = runBlocking {
    val deferred = async {
        delay(1000L)
        42
    }
    println("The answer is ${deferred.await()}")
}

runBlocking

阻塞当前线程直到协程完成，常用于测试或主函数。用于在 main函数中桥接阻塞与非阻塞代码。在作用域中启动的所有协程（如 launch {}）必须完成，runBlocking才会返回。

1
2
3
4
5
6
7
8
9
10
11
12
import kotlinx.coroutines.*

fun main() {
    runBlocking {
        launch {
            delay(1000)
            println("inside launch")
    	}
        
        println("Hello from runBlocking")
    }
}

上面的代码中， runBlocking { ... } 会阻塞当前线程，在当前线程中创建一个顶层协程（CoroutineScope），println("Hello from runBlocking") 是在 runBlocking 的协程中执行的（运行在主线程，但属于协程代码）。

withContext

在指定上下文运行挂起块，返回结果。

1
2
3
4
5
6
7
8
9
10
11
import kotlinx.coroutines.*

suspend fun fetchData(): String = withContext(Dispatchers.IO) {
    delay(1000L)
    "Data from network"
}

fun main() = runBlocking {
    val data = fetchData()
    println(data)
}

挂起函数

suspend

挂起函数用 suspend 关键字标记，只能从其他挂起函数或协程中调用，支持 delay 等挂起操作。 

1
2
3
4
5
6
7
8
9
10
import kotlinx.coroutines.*

suspend fun doSomething() {
    delay(1000L)
    println("Done something")
}

fun main() = runBlocking {
    doSomething()
}

协程作用域

Kotlin 协程作用域

# 协程作用域 是否结构化 是否阻塞线程 生命周期 典型用途 是否推荐
1 runBlocking ✅ 是 ✅ 是 直到 block 执行完毕 测试、main 函数入口 ✅（仅限测试/入口）
2 coroutineScope ✅ 是 ❌ 否 当前 suspend 函数结束 并发封装、函数内结构化并发 ✅ 推荐
3 supervisorScope ✅ 是 ❌ 否 当前 suspend 函数结束 容错并发（互不影响） ✅ 推荐
4 GlobalScope ❌ 否 ❌ 否 应用进程存活期间 后台任务（不推荐） ⚠️ 慎用（可能泄漏）
5 自定义 CoroutineScope(...) ✅/❌ 取决于 Job ❌ 否 由 Job 控制（手动 cancel） 封装类、组件内任务 ✅ 合理使用
7 actor / produce ✅ 是 ❌ 否 通道关闭或被取消 Channel 并发、消息队列 ✅ 特定场景推荐 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
import kotlinx.coroutines.*

fun main() = runBlocking {
    println("🔷 1. runBlocking started on [${Thread.currentThread().name}]")

    // 2. coroutineScope（结构化并发）
    coroutineScope {
        launch {
            delay(100)
            println("🔷 2. coroutineScope: child coroutine on [${Thread.currentThread().name}]")
        }
    }

    // 3. supervisorScope（不会因子协程失败而取消整个作用域）
    try {
        supervisorScope {
            launch {
                delay(200)
                println("🔷 3. supervisorScope: success task")
            }
            launch {
                delay(100)
                throw RuntimeException("🔷 3. supervisorScope: failed task")
            }
        }
    } catch (e: Exception) {
        println("🔴 Caught exception from supervisorScope: $e")
    }

    // 4. GlobalScope（不建议常用）
    val job = GlobalScope.launch {
        delay(300)
        println("🔷 4. GlobalScope coroutine on [${Thread.currentThread().name}]")
    }
    
    // 等待 GlobalScope 启动的协程完成
    // 如果不加这一行，上面的 GlobalScope 无法打印，因为 GlobalScope 启动的是一个全局线程，不是 runBlocking 的一部分，runBlocking 执行完就退出了，主线程结束，整个进程随之结束，导致 GlobalScope 中来不及打印就被强行终止了。
    job.join()

    // 5. 自定义 CoroutineScope
    val customScope = CoroutineScope(Job() + Dispatchers.Default)
    val customJob = customScope.launch {
        println("🔷 5. Custom CoroutineScope running on [${Thread.currentThread().name}]")
    }
    
    // 6. produce 示例（生产者管道）
    val numberProducer = produce {
        repeat(3) {
            delay(100)
            send(it * it)
        }
        close()
    }
    for (num in numberProducer) {
        println("🔷 6. produce: received $num")
    }

    // 7. actor 示例（带状态协程处理器）
    val sumActor = actor<Int> {
        var sum = 0
        for (msg in channel) {
            sum += msg
            println("🔷 7. actor: received $msg, current sum = $sum")
        }
    }

    listOf(1, 2, 3).forEach { sumActor.send(it) }
    sumActor.close()

    println("🕒 Waiting for GlobalScope & other coroutines...")
    delay(1000) // 等待其他 coroutine 执行完成

    // 记得手动 cancel 非结构化作用域
    customJob.join()
    customScope.cancel()
    println("✅ runBlocking completed.")
}

Android 扩展作用域

# Android Scope 是否结构化 生命周期绑定对象 典型用途 是否推荐
1 viewModelScope ✅ 是 ViewModel 生命周期 网络/IO/业务逻辑 ✅ 推荐
2 lifecycleScope ✅ 是 Activity / Fragment 生命周期 启动任务 / UI操作 ✅ 推荐
3 repeatOnLifecycle ✅ 是 生命周期状态变化（如 STARTED） 生命周期感知、数据收集（Flow） ✅ 推荐

MainActivity.kt

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
import android.os.Bundle
import androidx.activity.ComponentActivity
import androidx.activity.viewModels
import androidx.lifecycle.Lifecycle
import androidx.lifecycle.lifecycleScope
import androidx.lifecycle.repeatOnLifecycle
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.launch

class MainActivity : ComponentActivity() {

    private val viewModel by viewModels<MyViewModel>()

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)

        // ✅ 1. lifecycleScope：跟随 Activity 生命周期
        lifecycleScope.launch {
            delay(300)
            println("1️⃣ lifecycleScope → Thread: ${Thread.currentThread().name}")
        }

        // ✅ 2. repeatOnLifecycle：STARTED 时自动启动，STOPPED 时自动取消
        lifecycleScope.launch {
            repeatOnLifecycle(Lifecycle.State.STARTED) {
                viewModel.timerFlow.collect {
                    println("2️⃣ repeatOnLifecycle collect: $it")
                }
            }
        }

        // ✅ 3. viewModelScope（定义在 ViewModel 中）
        viewModel.loadData()
    }
}

MyViewModel.kt

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import androidx.lifecycle.ViewModel
import androidx.lifecycle.viewModelScope
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.launch

class MyViewModel : ViewModel() {

    // Flow 模拟定时器
    val timerFlow: Flow<Int> = flow {
        var counter = 0
        while (true) {
            emit(counter++)
            delay(1000)
        }
    }

    fun loadData() {
        viewModelScope.launch {
            delay(500)
            println("3️⃣ viewModelScope → Thread: ${Thread.currentThread().name}")
        }
    }
}

输出

1
2
3
4
5
6
1️⃣ lifecycleScope → Thread: main
3️⃣ viewModelScope → Thread: main
2️⃣ repeatOnLifecycle collect: 0
2️⃣ repeatOnLifecycle collect: 1
2️⃣ repeatOnLifecycle collect: 2
...

说明

  • lifecycleScope → 生命周期感知，自动取消协程（适合 UI 层数据加载等）

  • repeatOnLifecycle → 更细粒度的生命周期感知（可结合 Flow 实现 UI 活动期间自动收集）

  • viewModelScope → 随 ViewModel.onCleared() 自动取消，适合业务处理逻辑

使用场景

使用场景 推荐作用域
main 函数或测试 runBlocking
挂起函数内并发 coroutineScope
容错并发 supervisorScope
UI 层任务 viewModelScope / lifecycleScope
库或类封装内建 scope CoroutineScope(...)
Flow、通道数据源 produce / actor 内隐含作用域
后台不销毁任务 极少情况用 GlobalScope（通常不推荐）

协程上下文和调度器

调度器（如 Dispatchers.Main、Dispatchers.IO、Dispatchers.Default、Dispatchers.Unconfined）决定协程运行的线程。

1
2
3
4
launch(Dispatchers.Main) { }      // UI线程
launch(Dispatchers.IO) { }        // IO密集型
launch(Dispatchers.Default) { }   // CPU密集型
launch(Dispatchers.Unconfined) { } // 不固定线程，惯性调度
1
2
3
4
5
6
7
import kotlinx.coroutines.*

fun main() = runBlocking {
    launch(Dispatchers.IO) {
        println("Running on IO thread: ${Thread.currentThread().name}")
    }
}

Job 和 Deferred

Job

🔹 Job：表示一个可以被取消的协程任务

✅ 特点：

  • 代表一个正在运行的协程任务
  • 可以：
    • 取消（job.cancel()）
    • 等待其完成（job.join()）
  • 不返回结果值，只表示任务的生命周期

✅ 示例：

1
2
3
4
5
6
7
8
9
10
11
12
import kotlinx.coroutines.*

fun main() = runBlocking {
    val job = launch {
        delay(1000L)
        println("Job completed")
    }

    println("Waiting for job...")
    job.join() // 等待协程完成
    println("Main function continues")
}

Deferred

🔹 Deferred<T>：表示一个带返回值的协程任务（类似于 Future<T>）

✅ 特点：

  • 是 Job 的子类型
  • 表示一个异步计算的结果
  • 使用 async 创建，调用 .await() 获取结果
  • 也可以取消、检查是否完成

✅ 示例：

1
2
3
4
5
6
7
8
9
10
11
12
import kotlinx.coroutines.*

fun main() = runBlocking {
    val deferred: Deferred<Int> = async {
        delay(1000L)
        42
    }

    println("Doing other things...")
    val result = deferred.await() // 等待并获取返回值
    println("Result: $result")
}

🔍 Job 与 Deferred 的对比

属性/行为 Job Deferred<T>
是否返回结果值 ❌ 无 ✅ 有返回值 (T)
创建方式 launch { ... } async { ... }
获取结果方式 无 await()
取消任务 ✅ .cancel() ✅ .cancel()
等待完成 ✅ .join() ✅ .await()
是否继承自 Job ✅ 本身就是 ✅ 是 Job 的子接口
典型用途 启动独立任务 启动异步计算并获取值

✅ 使用建议

场景 使用类型 说明
启动任务但不关心结果 launch → Job 适合 UI 更新、日志、并发 fire-and-forget
启动异步任务并获取值 async → Deferred<T> 比如网络请求、计算、数据处理等

💡 补充知识点：async 的懒启动模式

1
2
3
4
5
6
val deferred = async(start = CoroutineStart.LAZY) {
    delay(1000)
    100
}
// ... 之后需要时调用 await 会启动
val result = deferred.await()

取消和超时

cancel

取消：用 cancel 取消协程。

1
2
3
4
5
6
7
8
9
10
11
12
13
import kotlinx.coroutines.*

fun main() = runBlocking {
    val job = launch {
        repeat(5) { i ->
            println("Coroutine executing $i")
            delay(500L)
        }
    }
    delay(1300L)
    job.cancel()
    println("Coroutine cancelled")
}

可响应取消的方式

1
2
3
4
5
suspend fun doWork() {
    while (isActive) {
        // 做一些任务
    }
}

withTimeout

超时：用 withTimeout 设置时间限制。 抛出 TimeoutCancellationException ❌

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import kotlinx.coroutines.*

fun main() = runBlocking {
    try {
        withTimeout(1300L) {
            repeat(5) { i ->
                println("Processing $i")
                delay(500L)
            }
        }
    } catch (e: TimeoutCancellationException) {
        println("Timed out")
    }
}

withTimeoutOrNull

返回 null，不抛异常 ✅

1
2
3
4
5
6
7
8
withTimeout(1000) {
    delay(2000) // 抛出 TimeoutCancellationException
}

val result = withTimeoutOrNull(1000) {
    delay(2000)
    "This won't return"
} ?: "Timeout!"

异常处理

使用 CoroutineExceptionHandler 处理异常（launch中使用）。

1
2
3
4
5
6
7
8
9
10
11
12
import kotlinx.coroutines.*

fun main() = runBlocking {
    // 在 Kotlin 中，_ 是一个 占位符变量名，表示“我不关心这个参数的值”。
    val handler = CoroutineExceptionHandler { _, exception ->
        println("Caught $exception")
    }
    launch(handler) {
        throw Exception("Error in coroutine")
    }
    delay(1000L)
}

使用 try-catch 处理异常。

1
2
3
4
5
6
7
launch {
    try {
        throw Exception("Error")
    } catch (e: Exception) {
        println("Caught: $e")
    }
}

协程组合与并发控制

并发执行任务 + 组合结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import kotlinx.coroutines.*

suspend fun getUser(): String {
    delay(1000) // 模拟耗时任务
    println("👤 getUser completed on [${Thread.currentThread().name}]")
    return "User123"
}

suspend fun getPosts(): List<String> {
    delay(800) // 模拟另一个耗时任务
    println("📝 getPosts completed on [${Thread.currentThread().name}]")
    return listOf("Post A", "Post B", "Post C")
}

fun main() = runBlocking {
    println("🚀 Running in [${Thread.currentThread().name}]")

    val userDeferred = async { getUser() }
    val postsDeferred = async { getPosts() }

    // to 是 Kotlin 中的一个非常常用的 中缀函数，用于创建键值对 (Pair)。
    // val (a, b) 是解构（destructuring），用多个变量同时接收一个对象（Pair、Triple、map 或数据类）的多个字段值。
    val (a, b) = userDeferred.await() to postsDeferred.await()
    println("✅ Combined result: $a + $b")
}

顺序执行多个挂起函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import kotlinx.coroutines.*

suspend fun getUser(): String {
    delay(1000)
    println("👤 getUser completed on [${Thread.currentThread().name}]")
    return "User456"
}

suspend fun getPosts(user: String): List<String> {
    delay(800)
    println("📝 getPosts for $user completed on [${Thread.currentThread().name}]")
    return listOf("Post X", "Post Y")
}

fun main() = runBlocking {
    println("🚀 Running in [${Thread.currentThread().name}]")

    val user = getUser()
    val posts = getPosts(user)

    println("✅ Result for $user: $posts")
}

区别说明

模式 是否并发 用法 适合场景
async + await ✅ 是 并发获取多个独立结果 多个不依赖关系的网络/API 请求
顺序调用 ❌ 否 一个接一个 后一个依赖前一个结果

结构化并发

结构化并发的核心：

  • 父协程自动管理子协程
  • 作用域退出时取消未完成子协程
  • 错误自动传播

coroutineScope

coroutineScope: 创建作用域，等待所有子协程完成。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import kotlinx.coroutines.*

suspend fun performTasks() = coroutineScope {
    launch {
        delay(1000L)
        println("Task 1")
    }
    launch {
        delay(2000L)
        println("Task 2")
    }
}

fun main() = runBlocking {
    performTasks()
    println("All tasks completed")
}

supervisorScope

supervisorScope: 类似，但子协程失败不影响其他。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import kotlinx.coroutines.*

suspend fun performTasks() = supervisorScope {
    launch {
        delay(1000L)
        throw Exception("Task 1 failed")
    }
    launch {
        delay(2000L)
        println("Task 2 completed")
    }
}

fun main() = runBlocking {
    try {
        performTasks()
    } catch (e: Exception) {
        println("Caught exception: $e")
    }
    println("Main function continues")
}

通道（Channels）

用于协程间通信，支持生产者-消费者模式。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun main() = runBlocking {
    val channel = Channel<Int>()
    
    launch {
        for (x in 1..5) channel.send(x * x)
        channel.close()
    }
    
    launch {
        for (y in channel) {
            println(y)
        }
	}
    
    println()
}

流（Flows）

Flow（协程版响应式流） 的特性：

  • 是冷流（调用 collect 才执行）
  • 自动挂起支持
  • 支持背压、变换、合并等操作符（如 map、filter、combine）

处理异步数据流。

1
2
3
4
5
6
7
8
9
10
11
12
13
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    flow {
        for (i in 1..3) {
            delay(100)
            emit(i)
        }
    }.collect { 
        value -> println(value) 
    }
}

选择表达式（Select Expression）

同时等待多个挂起操作，选择第一个可用结果。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
import kotlinx.coroutines.*
import kotlinx.coroutines.selects.*

suspend fun selectExample1(scope: CoroutineScope): String {
    return try {
        val deferred1 = scope.async {
            delay(1000L)
            "Result 1"
        }
        val deferred2 = scope.async {
            delay(500L)
            "Result 2"
        }

        select<String> {
            deferred1.onAwait { it }
            deferred2.onAwait { it }
        }
    } catch (e: Exception) {
        "Error occurred: ${e.message}"
    }
}

suspend fun selectExample2(): String = coroutineScope {
    // 表示从 coroutineScope block 中返回, 而不是返回外部函数，可省略
    return@coroutineScope try {
        val deferred1 = async {
            delay(1000L)
            "Result 1"
        }
        val deferred2 = async {
            delay(500L)
            "Result 2"
        }

        select<String> {
            deferred1.onAwait { it }
            deferred2.onAwait { it }
        }
    } catch (e: Exception) {
        "Error occurred: ${e.message}"
    }
}

fun main() {
    runBlocking {
        val result1 = selectExample1(this)
        println(result1)
        val result2 = selectExample2()
        println(result2)
    }
}

工具函数

yield

让出执行权，允许其他协程运行，适合协作式多任务。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import kotlinx.coroutines.*

fun main() = runBlocking {
    launch {
        repeat(5) {
            println("Coroutine 1: $it")
            yield()
        }
    }
    launch {
        repeat(5) {
            println("Coroutine 2: $it")
            yield()
        }
    }
}

输出：Coroutine 1 和 Coroutine 2 交替执行。

joinAll

等待多个 Job 完成。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import kotlinx.coroutines.*

fun main() {
    runBlocking {
        val job1 = launch {
            delay(1000L)
            println("Job 1")
        }

        val job2 = launch {
            delay(2000L)
            println("Job 2")
        }

        joinAll(job1, job2)
        println("All jobs completed")
    }
}

输出：Job 1、Job 2、All jobs completed。

awaitAll

等待多个 Deferred 完成，返回结果列表。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import kotlinx.coroutines.*

fun main() {
    runBlocking {
        val deferred1 = async {
            delay(1000L)
            1
        }

        val deferred2 = async {
            delay(2000L)
            2
        }

        val results = awaitAll(deferred1, deferred2)
        println(results)
    }
}

输出：Results: [1, 2]。

测试协程

使用 runBlockingTest 和 TestCoroutineDispatcher 测试协程，适合单元测试。 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import kotlinx.coroutines.test.*
import org.junit.Test

class CoroutineTest {
    @Test
    fun testCoroutine() = runBlockingTest {
        val deferred = async {
            delay(1000L)
            42
        }
        advanceTimeBy(1000L)
        assertEquals(42, deferred.await())
    }
}

实际应用场景

网络请求（Android）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import kotlinx.coroutines.*
import retrofit2.Retrofit
import retrofit2.converter.gson.GsonConverterFactory
import retrofit2.http.GET

interface ApiService {
    @GET("data")
    suspend fun getData(): String
}

class MainViewModel : ViewModel() {
    private val retrofit = Retrofit.Builder()
        .baseUrl("https://api.example.com/")
        .addConverterFactory(GsonConverterFactory.create())
        .build()
    private val apiService = retrofit.create(ApiService::class.java)

    fun fetchData() {
        viewModelScope.launch {
            try {
                val data = apiService.getData()
                println("Fetched data: $data")
            } catch (e: Exception) {
                println("Error: $e")
            }
        }
    }
}

数据库操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import androidx.room.*
import kotlinx.coroutines.*

@Entity
data class User(@PrimaryKey val id: Int, val name: String)

@Dao
interface UserDao {
    @Insert
    suspend fun insert(user: User)
}

class UserRepository(private val dao: UserDao) {
    suspend fun insertUser(user: User) = withContext(Dispatchers.IO) {
        dao.insert(user)
    }
}

fun main() = runBlocking {
    val repository = UserRepository(/* DAO instance */)
    launch { repository.insertUser(User(1, "John")) }
}

并发 API 调用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import kotlinx.coroutines.*

suspend fun fetchData1(): String = withContext(Dispatchers.IO) {
    delay(1000)
    "Data 1"
}

suspend fun fetchData2(): String = withContext(Dispatchers.IO) {
    delay(1500)
    "Data 2"
}

fun main() = runBlocking {
    val deferred1 = async { fetchData1() }
    val deferred2 = async { fetchData2() }
    val results = awaitAll(deferred1, deferred2)
    println("Results: $results")
}

超时处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import kotlinx.coroutines.*

suspend fun longRunningTask(): String = withContext(Dispatchers.IO) {
    delay(3000)
    "Done"
}

fun main() = runBlocking {
    try {
        withTimeout(2000L) {
            val result = longRunningTask()
            println(result)
        }
    } catch (e: TimeoutCancellationException) {
        println("Task timed out")
    }
}

参考资料