kotlin协程

发表于 分类于 本文字数： 6k
协程原理，基本使用

协程概念

kotlin官方：协程是线程的轻量级替代方案。它们可以在不阻塞系统资源的情况下挂起，并且资源利用率高，因此更适合细粒度的并发处理

个人拙见：Java线程和内核级线程的关系是一对一的，而Kotlin协程不直接与内核级线程产生关系，其是通过Java线程，间接的与操作系统内核级线程呈多对一或多对多的关系

Kotlin协程、Java线程、操作系统内核级线程之间的三种关系大致可做如下表述

  • 一对一：现代主流的Java实现中，Java创建的线程与操作系统内核线程一一对应，也因此，线程切换时开销较大
  • 多对多：在Kotlin中，当协程调度器指定为Dispatchers.Default或Dispatchers.IO时，协程运行在一个线程池中（多个协程在多个线程间复用和切换），线程池中的线程与内核线程一一对应，协程与线程池中的线程呈多对多的关系
  • 多对一：在Kotlin中，当调度器指定为Dispatchers.Main时，所有协程运行在一个线程中（如UI线程），此线程与内核线程相对应，协程与此线程之间呈多对一的关系

需要注意的是，如果协程的调度器指定为Dispatchers.Unconfined，协程将在调用它的线程中被启动，但挂起恢复后可能在任意线程继续执行，在此情况下协程与线程之间的关系无法确定

挂起和恢复

协程的挂起（Suspend）：挂起是指协程暂停执行，释放对当前占有线程的控制权，协程的挂起不会阻塞线程的执行，协程挂起后，线程可以继续执行其他任务
协程的恢复（Resume）：恢复是指被挂起的协程从之前暂停的地方继续执行，并且会恢复挂起时的执行状态和局部变量

如下代码中，协程A和协程B都在主线程运行，其输出表明，如果两个协程在同一个线程中，一个协程被挂起后，线程可以继续执行另一个协程，而不是阻塞

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
fun main() {
    runBlocking {
        launch {    // 协程A
            repeat(3) {
                println("${Thread.currentThread().name},Coroutine A")
                delay(1000) // 挂起协程，1秒后恢复
            }
        }

        launch {    // 协程B
            repeat(3) {
                println("${Thread.currentThread().name},Coroutine B")
                delay(1000) // 挂起协程，1秒后恢复
            }
        }
    }
    println("End")
}

// 有如下输出
// main,Coroutine A
// main,Coroutine B
// main,Coroutine A
// main,Coroutine B
// main,Coroutine A
// main,Coroutine B
// End

常见协程的挂起和恢复时机如下

挂起时机：

  • 显式挂起函数调用：delay()、yield()、await()、join()
  • 上下文切换：withContext()、withTimeout()
  • 协程间同步：Channel.receive()、Mutex.lock()
  • 异步操作转换：suspendCoroutine、callbackFlow

恢复时机：

  • 时间条件满足：delay()时间到，withTimeout()超时
  • 异步操作完成：网络响应、文件读取完成
  • 数据可用：Channel有数据，Flow发射新值
  • 协程完成：子协程执行完毕
  • 外部事件触发：回调被调用，信号量释放

挂起函数

通过suspend关键字修饰的函数称为挂起函数，其只能在协程或其他挂起函数中调用，挂起函数允许在执行过程中被挂起。常见一种说法“挂起函数被调用时，协程会被挂起，直到挂起函数执行完成后才会恢复继续执行”，但实际上挂起函数被调用时未必会挂起，是否挂起取决于函数内部是否存在挂起点，只有当协程执行到挂起点，并且满足挂起条件时，协程才会真正挂起

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
fun main() {
    runBlocking {
        launch {
            repeat(3) {
                exe("A")
            }
        }

        launch {
            repeat(3) {
                exe("B")
            }
        }
    }
    println("End")
}

suspend fun exe(flag: String) {
    println("${Thread.currentThread().name},Coroutine $flag")
    delay(1000) // 挂起点，协程在此处被挂起，1秒后恢复
}

常用协程作用域

CoroutineScope

注意：在CoroutineScope.kt文件下，以CoroutineScope命名的有一个接口和一个工厂函数，此处仅讨论工厂函数

1
CoroutineScope(context: CoroutineContext): CoroutineScope

CoroutineScope工厂函数提供了一种标准且安全的方式创建自定义作用域，其接收一个CoroutineContext类型的参数，创建并返回一个CoroutineScope接口类型的对象

1
2
3
4
5
6
// 源码中工厂函数的实现
public fun CoroutineScope(context: CoroutineContext): CoroutineScope =
    // 如果原context中已经包含Job，则直接使用该context创建ContextScope对象
    // 如果原context中未包含Job，则在原context基础上添加新的Job，并使用这个新的context创建ContextScope对象
    // ContextScope是CoroutineScope接口的实现类
    ContextScope(if (context[Job] != null) context else context + Job())

Job负责管理协程的生命周期（取消、等待、异常传播），CoroutineScope如果没有Job，那么从它启动的子协程将没有父Job，这些子协程将无法被统一取消或等待。可以通过如下的方式拿到工厂函数自动添加的Job

1
2
val scope = CoroutineScope(EmptyCoroutineContext)
val job = scope.coroutineContext[Job]

GlobalScope

全局作用域，伴随整个应用程序的生命周期，不会自动取消，除非程序结束或手动取消，也因此，在Android或前端等有明确生命周期的环境中，容易造成资源泄漏

GlobalScope是一个单例对象，从它启动的子协程没有父Job

1
2
3
4
5
public object GlobalScope : CoroutineScope {
    // coroutineContext返回空协程上下文，意味着子协程没有父Job
    override val coroutineContext: CoroutineContext
        get() = EmptyCoroutineContext
}

使用GlobalScope需要小心资源泄漏，如下代码，如果GlobalScope中的协程在Activity销毁后仍在运行，可能会导致内存泄漏

1
2
3
4
5
6
GlobalScope.launch(Dispatchers.Main) {
    // 模拟网络请求（实际应切换线程，这里简化）
    val result = fakeNetworkRequest()  // 假设耗时5秒
    // 5 秒后尝试更新 UI
    statusText.text = result  // 此时 Activity 可能已经销毁
}

MainScope

下面的代码本质上没有区别

1
2
3
MainScope()

CoroutineScope(Dispatchers.Main + SupervisorJob())

runBlocking

阻塞作用域，会阻塞当前线程直到内部所有协程执行完毕，一般仅用作调试

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
fun main() {
    runBlocking {
        launch {
            for (value in 1..2) {
                delay(10)
                println("coroutineA $value")
            }
        }

        launch {
            for (value in 1..2) {
                delay(10)
                println("coroutineB $value")
            }
        }
    }
    println("Hello, World!")
}

// 将会输出
// coroutineB 1
// coroutineA 1
// coroutineB 2
// coroutineA 2
// Hello, World!

实际上runBlocking有两个参数，第一个参数是协程上下文，一般不指定，默认使用EmptyCoroutineContext，第二个参数是一个供CoroutineScope对象调用的扩展函数，并且该函数的返回值也会成为runBlocking的返回值，所以也可以有如下写法

1
2
3
4
5
val blocking = runBlocking(EmptyCoroutineContext, object : (CoroutineScope) -> String {
    override fun invoke(scope: CoroutineScope): String {
        return "OK"
    }
})

coroutineScope

coroutineScope是一个挂起函数，用于在一个已有作用域中创建子作用域，该子作用域创建一个新的普通Job，而不使用父作用域的Job，因此该子作用域中并发执行多个子协程，遵循“要么全部成功，要么全部失败”的原则

1
2
3
4
5
6
7
8
9
10
suspend fun main() {
    coroutineScope {
        launch {
            // TODO 执行任务
        }
        launch {
            // TODO 执行任务
        }
    }
}

withContext

withContext允许在协程中临时切换上下文，如在Dispatchers.IO中执行耗时操作，在Dispatchers.Main中更新UI。它的返回值即为block函数的返回值，withContext会挂起当前协程，直到block函数执行完成后才会恢复继续执行

withContext不会像coroutineScope一样创建一个新的Job，它只是临时替换上下文，当前协程的Job不变

1
2
3
4
5
6
val data = withContext(Dispatchers.IO) {
    fetchData()
}
withContext(Dispatchers.Main) {
    binding.textView.text = data
}

supervisorScope & SupervisorJob

supervisorScope是一个挂起函数，所以只能在协程中调用，该函数允许在其中启动的子协程独立地处理异常，一个子协程的失败不会影响其他子协程的执行

1
2
3
4
5
6
val scope = CoroutineScope(Dispatchers.Main)
scope.launch {
    supervisorScope {
        // TODO 执行任务
    }
}

SupervisorJob是Job的子类，当CoroutineScope使用普通Job时，任何一个子协程失败，都会导致父Job被取消，进而连带所有子协程一起被取消，SupervisorJob使子协程的失败被局部化，只有它自己会被取消，其余部分继续独立运行

1
2
3
4
val scope = CoroutineScope(Dispatchers.Main + SupervisorJob())
scope.launch {
    // TODO 执行任务
}

协程构建器

launch

在现有协程作用域内部启动一个新协程，而不阻塞作用域的其余部分，当不需要结果或不想等待结果时，使用launch()在其他工作的同时运行任务

1
2
3
4
5
6
7
8
9
10
11
suspend fun performBackgroundWork() = coroutineScope {
    // 启动一个不阻塞作用域的协程
    this.launch {
        // 挂起以模拟后台工作
        delay(100.milliseconds)
        println("Sending notification in background")
    }

    // 主协程继续执行，而前一个协程处于挂起状态
    println("Scope continues")
}

async

在现有协程作用域内部启动一个并发计算，并返回一个Deferred句柄，该句柄代表最终结果，使用await()函数挂起代码直到结果准备就绪

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
suspend fun main() = withContext(Dispatchers.Default) { // this: CoroutineScope
    // 开始下载第一页
    val firstPage = this.async {
        delay(50.milliseconds)
        "First page"
    }

    // 并行开始下载第二页
    val secondPage = this.async {
        delay(100.milliseconds)
        "Second page"
    }

    // 等待两个结果并进行比较
    val pagesAreEqual = firstPage.await() == secondPage.await()
    println("Pages are equal: $pagesAreEqual")
}

参考