协程的互斥锁

我正在参与「启航方案」

在上一篇《协程的信号量》中，从一个简略的比如，咱们一窥 Kotlin 协程的信号量，即 Semaphore 的用法，也了解了合适它的场景。

今日来讲另一个同步的东西：Mutex，互斥锁

Mutex

名为 Mutex，和其他语言里的东西相同，便是一把同步锁，实现互斥机制，对临界区（比方共享资源）加以限制，用以保证多线程的共享资源安全。

和 Semaphore 相同，Kotlin 协程里的 Mutex，也是一个接口：

public interface Mutex {
    /**
     * 标志是否又锁
     */
    public val isLocked: Boolean
    /**
     * 尝试锁，假如已锁，则回来 false
     */
    public fun tryLock(owner: Any? = null): Boolean
    /**
     * 锁，假如 已锁，则挂起等待
     */
    public suspend fun lock(owner: Any? = null)
    /**
     * 检查 owner 是否锁，假如没有、或许锁的是其他 owner，回来 false
     */
    public fun holdsLock(owner: Any): Boolean
    /**
     * 解锁
     */
    public fun unlock(owner: Any? = null)
}

值得注意的是，一切「锁」办法，都带了一个参数，名为 owner。它的效果就像一个 key 相同，决定了关注的是哪把锁。换句话说：一个 Mutex 能够锁不同的 owner。一旦运用了 owner，但又不符合锁的状况要求而调用了特定的办法，可能会抛反常。

相似信号量，Mutex 也供给了办法获取实例：

public fun Mutex(locked: Boolean = false): Mutex =
    MutexImpl(locked)

能够通过参数 locked，来设置初始状况是否锁住。

然后呢，同样也有一个 withXXX 东西函数封装，帮咱们处理了「锁/解锁」的操作配对问题：

public suspend inline fun <T> Mutex.withLock(owner: Any? = null, action: () -> T): T {
    contract { 
        callsInPlace(action, InvocationKind.EXACTLY_ONCE)
    }
    lock(owner) // 锁住目标owner
    try {
        return action() // 执行使命
    } finally {
        unlock(owner) // 解锁
    }
}

案例

看过之前信号量的文章，现在再看 Mutex 的，就能轻车熟路般地运用它了。借用之前的比如，改造一下使命实现：

private val mutex = Mutex()
private suspend fun mutexTask(name: String, owner: Any? = null) {
    println("$name locking")
    mutex.withLock(owner) {
        println("$name locked")
        delay(1000L)
        println("$name unlocked")
    }
}
private suspend fun tryMutexTask(name: String, owner: Any? = null) {
    println("try $name locking")
    if (mutex.tryLock(owner)) {
        println("try $name locked")
        delay(1500)
        println("$name unlocked")
        mutex.unlock(owner)
    } else {
        println("try $name failed")
    }
}

1

同样的，先看看一般 lock 下的运用：

for (i in 0 until 3) {
    GlobalScope.launch {
        mutexTask("task-$i")
    }
}
println("all posted")
GlobalScope.launch {
    while (true) {
        print(".")
        delay(200)
    }
}
delay(5_000L)
println("done")

成果 ：

all posted
task-1 locking
task-1 locked
task-2 locking
task-0 locking
.....task-1 unlocked
task-2 locked
.....task-2 unlocked
task-0 locked
.....task-0 unlocked
..........done

三个使命都尝试锁，但只有使命 1 成功；使命 1 完成后解锁，这时，使命 2 成功锁住并执行使命。后边的使命 0 与此相似。

能够看出，Mutex 便是一种「独占式」的保护门。

2

咱们再来看看「尝试锁」是怎么样的：

GlobalScope.launch {
    tryMutexTask("before")
}
for (i in 0 until 3) {
    GlobalScope.launch {
        mutexTask("task-$i")
    }
}
println("all posted")
GlobalScope.launch {
    tryMutexTask("after")
}
// ...

成果：

task-0 locking
task-0 locked
all posted
task-1 locking
task-2 locking
try before locking
try before failed
try after locking
try after failed
.....task-0 unlocked
task-2 locked
.....task-2 unlocked
task-1 locked
.....task-1 unlocked
..........done

这个成果完美体现了多线程的不确定性：首先获得锁的，是使命0，而不是 before。这就导致 before 的 try 失利。其他流程和前面实验 1 相同了。

再运行一遍：

try before locking
try before locked
task-0 locking
task-1 locking
all posted
task-2 locking
try after locking
try after failed
........before unlocked
task-0 locked
.....task-0 unlocked
task-1 locked
.....task-1 unlocked
task-2 locked
.....task-2 unlocked
..done

嗯，这次 before 先锁住了，try 成功。于是乎，后边的三个使命全都锁挂起，after 是 try 锁，直接失利回来。然后同样的，后边便是「锁住、解锁」的成对序列了。

3

前面接口说明有提到，假如设置了 owner 的锁，假如调用不合时宜，将抛反常。

比方下面：

for (i in 0 until 3) {
    GlobalScope.launch {
        mutexTask("task-$i", "try")
    }
}

预想的溃散来了：

all posted
task-0 locking
task-1 locking
task-2 locking
.task-0 locked
Exception in thread "DefaultDispatcher-worker-2" Exception in thread "DefaultDispatcher-worker-3" java.lang.IllegalStateException: Already locked by try
    at kotlinx.coroutines.sync.MutexImpl.lockSuspend(Mutex.kt:208)
    at kotlinx.coroutines.sync.MutexImpl.lock(Mutex.kt:186)
    at coroutine.SyncKt.mutexTask(sync.kt:115)
    at coroutine.SyncKt.access$mutexTask(sync.kt:1)
    at coroutine.SyncKt$main$1$1.invokeSuspend(sync.kt:86)
    at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
    at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
    at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:570)
    at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:750)
    at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:677)
    at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:664)
    Suppressed: kotlinx.coroutines.DiagnosticCoroutineContextException: [StandaloneCoroutine{Cancelling}@663fba34, Dispatchers.Default]
java.lang.IllegalStateException: Already locked by try
    at kotlinx.coroutines.sync.MutexImpl.lockSuspend(Mutex.kt:208)
    at kotlinx.coroutines.sync.MutexImpl.lock(Mutex.kt:186)
    at coroutine.SyncKt.mutexTask(sync.kt:115)
    at coroutine.SyncKt.access$mutexTask(sync.kt:1)
    at coroutine.SyncKt$main$1$1.invokeSuspend(sync.kt:86)
    at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
    at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
    at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:570)
    at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:750)
    at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:677)
    at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:664)
    Suppressed: kotlinx.coroutines.DiagnosticCoroutineContextException: [StandaloneCoroutine{Cancelling}@702043ed, Dispatchers.Default]
....task-0 unlocked

第一个 owner 为 「try」的锁能够，后边再来就抛反常了。

当然，改改名就能够处理：

for (i in 0 until 3) {
    GlobalScope.launch {
        mutexTask("task-$i", "try$i")
    }
}

这样每次的 owner 就都不同了。

再谈信号量 Semaphore

看到这儿，有没有发现，Mutex 的效果，甚至在运用上，都和信号量 Semaphore 很相似啊？

没错，其实考虑一下就清楚了，对于信号量答应数设置为 1 的 Semaphore，不便是一个 Mutex 吗？ 每运用一个信号量，就没有剩余的存在，其他恳求方必须等这一个开释才能获取，妥妥地便是一个互斥锁。

小结

相较于信号量，Mutex 更为简略。不过呢，运用的出错概率也增加了，因为它动不动就有可能来个溃散，还是小心运用为好。