Kotlin 协程基础十 —— 协作、互斥锁与共享变量

Kotlin 协程基础系列：

Kotlin 协程基础一 —— 总体知识概述

Kotlin 协程基础二 —— 结构化并发（一）

Kotlin 协程基础三 —— 结构化并发（二）

Kotlin 协程基础四 —— CoroutineScope 与 CoroutineContext

Kotlin 协程基础五 —— Channel

Kotlin 协程基础六 —— Flow

Kotlin 协程基础七 —— Flow 操作符（一）

Kotlin 协程基础八 —— Flow 操作符（二）

Kotlin 协程基础九 —— SharedFlow 与 StateFlow

Kotlin 协程基础十 —— 协作、互斥锁与共享变量

本节将介绍在协程间如果有先后执行、互相等待的需求时，应该怎样去处理这种等待和协作的工作。更会与 Java 的线程的协作工作对比，从而说明，在线程中通常不太简单的协作操作，在协程中很容易实现。

1、协程间的协作与等待

从运行角度来看，协程天生就是并行的，不论是对同等级的协程、父子协程还是毫无关系的协程。假如我们需要让协程互相等待，希望在协程的执行过程中可以停住，等待别的协程执行完毕，可以使用 Job 的 join() 或 Deferred 的 await()。

线程对于这种互相等待的需求可以通过 Thread 的 join()，还有 Future 和 CompletableFuture 以及 CountDownLatch。

CountDownLatch 适用于一个线程等待多个线程：

fun main() = runBlocking<Unit> {
    // countdown 译为倒计时，latch 是门闩、插销，组合起来就是用于倒计时的插销
    val countDownLatch = CountDownLatch(2)
    thread {
        // await() 会在 CountDownLatch 内的 count 减到 0 时结束等待
        countDownLatch.await()
        println("Count in CountDownLatch is 0 now,I'm free!")
    }

    thread {
        sleep(1000)
        // countDown() 会调用原子操作让 CountDownLatch 内的 count 减 1
        countDownLatch.countDown()
        println("Invoke countDown,count: ${countDownLatch.count}")
    }

    thread {
        sleep(2000)
        countDownLatch.countDown()
        println("Invoke countDown,count: ${countDownLatch.count}")
    }
}

运行结果：

Invoke countDown,count: 1
Count in CountDownLatch is 0 now,I'm free!
Invoke countDown,count: 0

修改 CountDownLatch 构造方法的 count 参数就可以修改要等待的线程数量，对于这种一个等待多个的业务需求，在协程中也可以用 join() 来做：

fun main() = runBlocking<Unit> {
    // 两个前置任务
    val preJob1 = launch {
        delay(1000)
    }

    val preJob2 = launch {
        delay(2000)
    }

    // 此协程需要等待两个协程执行之后再运行自己的内容
    launch {
        preJob1.join()
        preJob2.join()
        // 等待完前置任务，再做自己的事...
    }
}

实际上线程里也可以这么做，只不过因为线程本身的结构化管理比较麻烦，所以在正式的项目里很少真正的这么写。但因为协程可以结构化取消，因此它的 join() 比线程的 join() 更实用，在正式项目里的应用也较多。

其实，用 Channel 也能实现类似 CountDownLatch 那种，不指定具体等待哪些协程，只等待固定的次数的效果：

private fun channelSample() = runBlocking<Unit> {
    // 指定 Channel 的容量为 2
    val channel = Channel<Unit>(2)

    // 由于要等待两次发送数据才能继续执行后续代码，因此要 repeat(2) 接收
    launch {
        repeat(2) {
            channel.receive()
        }
    }
    
    launch {
        delay(1000)
        channel.send(Unit)
    }

    launch {
        delay(2000)
        channel.send(Unit)
    }
}

通过两个简单的例子可以发现，线程中有些复杂、比较底层、不太容易使用的协作和等待 API，在协程中的对应/等价 API 难度要大大降低。

2、select()：先到先得

select() 会在内部开启多线竞赛，谁最快就用谁。

onJoin() 是仅限于在 select 代码块中才能调用的函数，它是一个监听注册，会监听 Job 的结束，不论 Job 是正常结束还是被取消，在其结束时都会回调执行 onJoin() 大括号的内容，并且大括号的返回值会作为 select() 的返回值：

fun main() = runBlocking<Unit> {
    val scope = CoroutineScope(EmptyCoroutineContext)

    val job1 = scope.launch {
        delay(1000)
        println("job1 done")
    }

    val job2 = scope.launch {
        delay(2000)
        println("job2 done")
    }

    val job = scope.launch {
        val result = select {
            // select 只执行最先结束的 onJoin 回调
            job1.onJoin {
                1
            }

            job2.onJoin {
                2
            }
        }
        println("result: $result")
    }
    joinAll(job, job1, job2)
}

运行结果：

job1 done
result: 1
job2 done

结果能看出，select() 只执行了最先结束的 job1 的 onJoin，没有执行 job2 的。

与 Job 的 onJoin() 功能类似的还有：Deferred 的 onAwait()、Channel 的 onSend()、onReceive() 以及 onReceiveCatching()。此外还有一个特殊的函数 onTimeout()，如果 select() 内所有的监听回调都没有在 onTimeout() 设置的超时时间内完成，那么就由 onTimeout() 作为 select() 的返回值：

@OptIn(ExperimentalCoroutinesApi::class)
fun main() = runBlocking<Unit> {
    val scope = CoroutineScope(EmptyCoroutineContext)

    val job1 = scope.launch {
        delay(1000)
        println("job1 done")
    }

    val deferred = scope.async {
        delay(2000)
        println("deferred done")
    }

    val channel = Channel<String>()

    val job = scope.launch {
        val result = select {
            // select 只执行最先结束的 onJoin 回调
            job1.onJoin {
                1
            }

            deferred.onAwait {
                2
            }

            channel.onSend("haha") {}
            /*channel.onReceive {}
            channel.onReceiveCatching {}*/

            onTimeout(500) {
                "Timeout!"
            }
        }
        println("result: $result")
    }
    joinAll(job, job1)
}

运行结果：

result: Timeout!
job1 done

3、互斥锁和共享变量

在遇到一个不太好理解的知识点时，我们还是先说线程，再引入到协程中。

线程中有一个术语叫竞态条件，或者说竞争条件，英文是 race condition。这个词的含义比较广，在 Java 和 Kotlin 这种高级编程语言中，它指的是多个线程访问共享资源时，由于缺乏并发控制，导致资源的访问顺序不受控，进而导致出现错误的结果的条件。

在 Kotlin 中，仍然可以使用我们在 Java 中熟知的 synchronized 和 Lock 这两种锁机制来保证共享资源的线程安全，也提供了新的选项，下面我们来说一说。

3.1 @Synchronized

Kotlin 中没有 synchronized 关键字，代替它的是 @Synchronized 注解。对于方法而言，使用 @Synchronized 注解的作用与 Java 中使用 synchronized 关键字修饰方法的作用是一样的。被 @Synchronized 注解标记的方法不能同时被多个线程（注意，不是协程）执行。

而 Java 中 synchronized 代码块在 Kotlin 中被 synchronized 函数代替了：

fun main() = runBlocking<Unit> {
    var number = 0
    val lock = Any()

    val thread1 = thread {
        repeat(100_000) {
            synchronized(lock) {
                number++
            }
        }
    }

    val thread2 = thread {
        repeat(100_000) {
            synchronized(lock) {
                number--
            }
        }
    }

    thread1.join()
    thread2.join()
    println("result: $number") // 输出 0
}

同样的代码结构也可以用在协程中：

fun main() = runBlocking<Unit> {
    var number = 0
    val lock = Any()
    val scope = CoroutineScope(EmptyCoroutineContext)
    val job1 = scope.launch {
        repeat(100_000) {
            synchronized(lock) {
                number++
            }
        }
    }

    val job2 = scope.launch {
        repeat(100_000) {
            synchronized(lock) {
                number--
            }
        }
    }

    job1.join()
    job2.join()
    println("result: $number")
}

synchronized() 仍然掐住的是线程，确切的说是掐住了执行 synchronized() 所在的协程的线程。虽然这样做有点浪费，因为不止掐住了协程，连运行该协程代码的线程都被掐住了，但确实实现了共享资源的线程安全，而且 synchronized() 本来也是针对线程的，只不过从协程的角度看，如果可以只掐住协程，不影响运行该协程代码的线程就更好了。

这个区别就好像 delay() 与 sleep() 一样。协程的 delay() 只会挂起当前的协程，但是不会影响其所在的线程；而 sleep() 是让整个线程休眠。因此在协程中，为了不影响整个线程，我们通常都是使用 delay() 仅作用于当前协程，而不会使用 sleep() 为了让协程挂起而影响到整个线程的运行。下一节要讲的 Mutex 就可以解决这个问题。

Lock 的用法也大致相同，这里不多赘述。

3.2 Mutex

Mutex 是计算机领域的专属词汇，全称是 mutual exclusion，即互斥。Kotlin 提供的 Mutex 是基于协程的、挂起式的，不同于前面两个是基于线程的、阻塞式的。Mutex 是协程自己的实现，它不卡线程，性能更好，使用也很方便：

fun main() = runBlocking<Unit> {
    var number = 0
    val mutex = Mutex()
    val scope = CoroutineScope(EmptyCoroutineContext)
    val job1 = scope.launch {
        repeat(100_000) {
            try {
                mutex.lock()
                number++
            } finally {
                mutex.unlock()
            }
        }
    }

    val job2 = scope.launch {
        repeat(100_000) {
            mutex.withLock {
                number--
            }
        }
    }

    job1.join()
    job2.join()
    println("result: $number")
}

job1 内使用的是常规用法，在操作共享变量前用 lock() 加锁，在 finally 代码块中解锁。job2 内使用的是简便写法，withLock() 将代码块内的代码放入 try 中执行，在 finally 中用 unlock() 解锁：

@OptIn(ExperimentalContracts::class)
public suspend inline fun <T> Mutex.withLock(owner: Any? = null, action: () -> T): T {
    contract {
        callsInPlace(action, InvocationKind.EXACTLY_ONCE)
    }
    lock(owner)
    return try {
        action()
    } finally {
        unlock(owner)
    }
}

Mutex 的优势是性能，但由于它是基于协程的，因此只能在协程中使用。所以，如果只在协程中使用共享资源，那么就用 Mutex，如果需要在线程中使用，就要用上一节说的 synchronized 与 Lock。

3.3 Semaphore

Java 还有一个 Semaphore，信号量，一个可以被多个线程持有的锁。你可以在它的构造方法中指定它最多可以被几个线程持有，如果有多余指定数量的线程去获取 Semaphore 就会陷入等待。获取锁用 acquire()，释放锁用 release()。

由于共享变量是只要有两个线程同时访问就会导致出错了，因此允许多个线程持有的 Semaphore 并不能用于解决竞态条件的问题，它是用来做性能控制的。你可以用它来实现类似线程池的功能，只不过你实现出来的是自己定制的对象池：同一时间最多只有多少个对象同时在做事，满了之后如果再来新对象就得等着，直到有新的坑让出来，这些新对象才能开始做事。

Kotlin 提供了一个 Semaphore 的协程版本，就叫 Semaphore，定位与 Java 的 Semaphore 相同，只不过是协程版本。

3.4 其他 API

在传统的线程系统里，还有一组典型的 API：wait()、notify()、notifyAll()。它们三个属于更底层的 API，在线程系统里，它既能实现互斥锁，也能实现线程之间相互等待的功能。但事实上，这些年已经基本没人再用这组函数了。因为 synchronized 关键字与 Lock 的推出，已经基本上完全替代了它们，而且它们用起来也很麻烦，所以现在没人用。正因如此，协程没有推出与它们类似的 API。

AtomicInteger 与 CopyOnWriteArrayList 等等也可以在协程中使用。虽然它们是针对线程的，但是卡住线程的同时一定把协程也卡住了。所以在协程里也可以无风险地使用。

此外，volatile 与 transient 也可以在协程中使用，只不过不再是关键字，而是注解。

4、ThreadLocal

ThreadLocal 是线程的局部变量，即该变量在每个线程都是独立的，从不同的线程中访问该变量，这些线程对变量的值的读写都是相互独立的，对每个线程都有独立的副本。

ThreadLocal 是用来干嘛的？它的定位就像它的名字一样，就是针对线程的局部变量。Java 变量按照作用域由小到大可以划分为局部变量（方法内）、成员变量（类内）、静态变量（全局），ThreadLocal 是一种介于局部变量和静态变量之间的一种变量，范围比方法大，比静态全局小，只在当前线程范围内有效。

ThreadLocal 是对 Java 线程一个很关键的能力补充。前面提过，协程相对线程的一大优势就是线程不具备结构化管理的能力，而协程结构化管理的能力相当强大。线程不具备结构化管理的能力，但我们开发时是有结构化管理的需求的，这时就要用 ThreadLocal。有了 ThreadLocal 之后，在同一个线程里执行的多个方法之间就可以共享变量了，且该共享变量只针对当前线程有效，跨线程时还是独立的。因此 ThreadLocal 通常会作为静态变量存在。

ThreadLocal 在协程中的等价物是什么？有什么东西是跨方法的、针对协程的局部变量吗？CoroutineContext 就是协程里的 ThreadLocal。

本来，由于协程是具备结构化管理能力的，你完全不需要在协程内使用 ThreadLocal。但是开发过程中，免不了与 Java 代码进行协作，如果想在协程代码里访问老代码里的 ThreadLocal 对象，是不能像如下这样直接使用的：

val kotlinLocalString = ThreadLocal<String>()
fun main() = runBlocking<Unit> {
    val scope = CoroutineScope(EmptyCoroutineContext)
    val job = scope.launch {
        kotlinLocalString.set("Test")
        delay(1000)
        println(kotlinLocalString.get())
    }
    job.join()
}

kotlinLocalString 的 get() 拿到的值一定是 set() 设置的值吗？不一定！因为虽然协程没变，但是执行协程代码的线程有可能改变了，delay() 的时候线程被让出，可能会去执行其他协程的代码。等 delay() 结束继续执行下面代码的时候，有可能就不是在刚才的线程中执行了。因为协程只能保证在执行挂起函数之后依然运行在刚才的 ContinuationInterceptor 所管理的某一个线程池上，不能保证同一个线程。

因此 ThreadLocal 不能在协程中直接使用，因为它的效果在协程中变得不可靠了。怎么办？用 asContextElement() 把 ThreadLocal 转换成 CoroutineContext：

val kotlinLocalString = ThreadLocal<String>()
fun main() = runBlocking<Unit> {
    val scope = CoroutineScope(EmptyCoroutineContext)
    val job = scope.launch {
        val stringContext = kotlinLocalString.asContextElement("Test")
        withContext(stringContext) {
            delay(1000)
            println(kotlinLocalString.get())
        }
    }
    job.join()
}

asContextElement() 是 ThreadLocal 的扩展函数，它会把参数里的值封装到返回值的 ThreadLocalElement 中。再将结果填到 withContext() 的参数中，包住获取 ThreadLocal 值的代码，这时候里面的 ThreadLocal 就是对协程兼容的了。不管里面怎么切协程，只要没出协程，它的值都会被保持住。