如何定义两个协程间的依赖？让协程2等待SharedFlow订阅完成

问题场景 #

现有如下代码，协程1通过viewModelScope启动并无限收集SharedFlow，协程2需要在协程1完成SharedFlow订阅后，再执行myService.initialize()，避免初始化后发射的值被错过：

init {
     // 协程1
     // 由于该收集操作会无限运行，因此需要放在单独的协程中
    viewModelScope.launch {
        myService.someSharedFlow.collect {
            // 处理值
        }
    }

    // 协程2
    viewModelScope.launch {
        // 此代码不得在协程1中的SharedFlow订阅完成前执行
        // 以确保不会错过任何发射的值
        withContext(Dispatchers.IO) {
            myService.initialize() // 初始化后会通过Flow发送一个值
        }
    }
}

解决方案 #

方案1：利用SharedFlow的subscriptionCount实现同步 #

SharedFlow自带subscriptionCount属性，它会实时发射当前订阅者的数量。我们可以等待这个计数变为1（表示协程1的订阅已完成），再执行协程2的初始化逻辑：

init {
    // 协程1
    viewModelScope.launch {
        myService.someSharedFlow.collect {
            // 处理值
        }
    }

    // 协程2
    viewModelScope.launch {
        // 等待subscriptionCount达到1，确认已有活跃订阅者
        myService.someSharedFlow.subscriptionCount
            .filter { it >= 1 }
            .first() // 取第一个满足条件的值后结束流

        withContext(Dispatchers.IO) {
            myService.initialize()
        }
    }
}

注意：如果someSharedFlow是MutableSharedFlow，需确保它的replay、extraBufferCapacity参数配置符合需求，避免初始化时的发射值被丢弃。

方案2：使用CompletableDeferred手动同步 #

创建一个CompletableDeferred对象，在协程1的订阅建立时标记完成，协程2等待这个信号后再执行初始化：

init {
    val subscriptionReady = CompletableDeferred<Unit>()

    // 协程1
    viewModelScope.launch {
        myService.someSharedFlow
            .onSubscription {
                // 订阅建立时立即标记完成
                subscriptionReady.complete(Unit)
            }
            .collect {
                // 处理值
            }
    }

    // 协程2
    viewModelScope.launch {
        // 等待订阅完成信号
        subscriptionReady.await()

        withContext(Dispatchers.IO) {
            myService.initialize()
        }
    }
}

这个方案更灵活，onSubscription能精准捕捉订阅建立的时机，适合需要严格控制同步节点的场景。

内容的提问来源于stack exchange，提问作者me.at.coding