Kotlin协程中非阻塞懒计算值的可重置线程安全缓存实现方案

Kotlin标准库及协程库目前没有内置的对应组件，但可以通过非常短的代码实现满足所有需求的版本，有两种常用实现方案：

方案1：无锁CAS实现（推荐） #

不需要额外依赖，性能更高，reset方法支持非协程场景调用：

import java.util.concurrent.atomic.AtomicReference
import kotlinx.coroutines.Deferred
import kotlinx.coroutines.CompletableDeferred

class CachedComputation<T>(private val compute: suspend () -> T) {
    private val deferredRef = AtomicReference<Deferred<T>?>(null)

    suspend fun get(): T {
        while (true) {
            val current = deferredRef.get()
            when {
                // 无缓存或缓存已失效，启动新计算
                current == null || current.isCancelled -> {
                    val newDeferred = CompletableDeferred<T>()
                    // CAS保证只有一个线程能成功设置新的Deferred
                    if (deferredRef.compareAndSet(current, newDeferred)) {
                        runCatching { compute() }
                            .onSuccess { newDeferred.complete(it) }
                            .onFailure { newDeferred.completeExceptionally(it) }
                        return newDeferred.await()
                    }
                }
                // 已有正在运行/已完成的任务，直接等待结果
                else -> return current.await()
            }
        }
    }

    fun reset() {
        // 清空缓存并取消正在运行的任务
        deferredRef.getAndSet(null)?.cancel()
    }
}

特性说明： #

• 完全满足所有需求：已完成直接返回缓存、运行中所有调用者共享结果等待、无缓存时自动启动计算
• 无GlobalScope依赖，compute逻辑完全继承第一个调用get()的协程上下文
• 线程安全，无锁CAS操作性能优于同步锁
• 自动处理异常：计算抛出异常时所有等待的调用者都会收到对应异常，下次调用get()会自动重启计算
• reset方法为普通函数，可在任意线程直接调用，会主动取消正在运行的计算任务

方案2：Mutex实现（逻辑更直观） #

如果不想处理CAS循环，可以用协程轻量级锁Mutex实现，逻辑更易读：

import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.Deferred
import kotlinx.coroutines.CompletableDeferred

class CachedComputation<T>(private val compute: suspend () -> T) {
    private val mutex = Mutex()
    private var deferred: Deferred<T>? = null

    suspend fun get(): T = mutex.withLock {
        val current = deferred
        when {
            current == null || current.isCancelled -> {
                CompletableDeferred<T>().also { d ->
                    deferred = d
                    runCatching { compute() }
                        .onSuccess { d.complete(it) }
                        .onFailure { d.completeExceptionally(it) }
                }.await()
            }
            else -> current.await()
        }
    }

    suspend fun reset() = mutex.withLock {
        deferred?.cancel()
        deferred = null
    }
}

特性说明： #

• 逻辑更简单易维护，适合不需要高频调用reset的场景
• 所有特性和方案1一致，仅reset方法需要在协程上下文调用

内容的提问来源于stack exchange，提问作者Marcin