Kotlin多协程Scope运行正常，同Scope下为何失效？

我尝试从DataStore中读取数据填充对应变量，希望仅获取一次数据，因此在1秒后取消Job以避免Flow持续推送更新。但以下两种基于同一Coroutine Scope的实现均无法正常工作：

• 在单个launch中依次collect多个DataStore Flow，仅第一个变量被赋值；
• 在同一Scope下启动多个launch分别collect不同Flow，同样无法正常填充所有变量；
  只有为每个Flow创建独立Coroutine Scope的实现可以正常运行。想了解三种实现差异的原因。

无效实现(1) #

suspend fun setupDataStore(context: Context) {
    tempDataStore = TempDataStore(context)
    val job = Job()
    val scope = CoroutineScope(job + Dispatchers.IO)
    scope.launch {
        tempDataStore.getDieOne.collect {
            die1.value = it!!.toInt()
        }
        tempDataStore.getDisplayText.collect {
            displayText.value = it!!
        }
        tempDataStore.getDieTwo.collect {
            die2.value = it!!.toInt()
        }
    }
    delay(1000L)
    job.cancel()
}

无效实现(2) #

suspend fun setupDataStore(context: Context) {
    tempDataStore = TempDataStore(context)
    val job = Job()
    val scope = CoroutineScope(job + Dispatchers.IO)
    scope.launch {
        tempDataStore.getDieOne.collect {
            die1.value = it!!.toInt()
        }
    }
    scope.launch {
        tempDataStore.getDisplayText.collect {
            displayText.value = it!!
        }
    }
    scope.launch {
        tempDataStore.getDieTwo.collect {
            die2.value = it!!.toInt()
        }
    }
    delay(1000L)
    job.cancel()
}

有效实现(3) #

suspend fun setupDataStore(context: Context) {
     tempDataStore = TempDataStore(context)
     val job = Job()
     val job2 = Job()
     val job3 = Job()
     val scope = CoroutineScope(job + Dispatchers.IO)
     val scope2 = CoroutineScope(job2 + Dispatchers.IO)
     val scope3 = CoroutineScope(job3 + Dispatchers.IO)
     scope.launch {
          tempDataStore.getDieOne.collect {
             die1.value = it!!.toInt()
         }
     }
     scope2.launch {
          tempDataStore.getDisplayText.collect {
             displayText.value = it!!
         }
     }
     scope3.launch {
          tempDataStore.getDieTwo.collect {
              die2.value = it!!.toInt()
         }
     }
     delay(1000L)
     job.cancel()
     job2.cancel()
     job3.cancel()
}

TempDataStore类 #

class TempDataStore(private val context: Context) {

    companion object{
        private val Context.dataStore by preferencesDataStore(name = "TempDataStore")
        val DISPLAY_TEXT_KEY = stringPreferencesKey("display_text")
        val DIE_ONE = stringPreferencesKey("die_one")
        val DIE_TWO = stringPreferencesKey("die_two")
    }

    val getDisplayText: Flow<String?> = context.dataStore.data
        .map { preferences ->
            preferences[DISPLAY_TEXT_KEY] ?: "Roll to start!"
        }

    suspend fun saveDisplayText(text: String) {
        context.dataStore.edit { preferences ->
            preferences[DISPLAY_TEXT_KEY] = text
        }
    }

    val getDieOne: Flow<String?> = context.dataStore.data
        .map { preferences ->
            preferences[DIE_ONE] ?: "1"
        }

    suspend fun saveDieOne(dieOne: Int) {
        context.dataStore.edit { preferences ->
            preferences[DIE_ONE] = dieOne.toString()
        }
    }

    val getDieTwo: Flow<String?> = context.dataStore.data
        .map { preferences ->
            preferences[DIE_TWO] ?: "2"
        }

    suspend fun saveDieTwo(dieTwo: Int) {
        context.dataStore.edit { preferences ->
            preferences[DIE_TWO] = dieTwo.toString()
        }
    }

    suspend fun resetDataStore() {
        context.dataStore.edit { preferences ->
            preferences.clear()
        }
    }
}

Composable调用代码 #

LaunchedEffect(true) {
   sharedViewModel.setRoles()
   sharedViewModel.saveChanges()
   sharedViewModel.setupDataStore(context)
}

三种实现的差异原因 #

1. 无效实现(1)的问题 #

collect是挂起函数，会一直阻塞当前协程直到Flow被取消或终止。在单个launch里依次调用三个collect时，第一个tempDataStore.getDieOne.collect会持续占用该协程，后面两个collect完全没有执行机会，因此只有die1变量能被赋值，另外两个变量永远不会被处理。

2. 无效实现(2)的问题 #

同一Coroutine Scope下的所有launch共享同一个父Job。当调用delay(1000L)后取消父Job时，部分Flow可能还没完成第一次数据发射就被中断。此外，Preferences DataStore内部的读写操作是串行处理的（保证线程安全），多个collect发起的读取请求会排队执行，1秒延迟可能不足以让所有读取完成，导致部分变量未被赋值就被取消。

3. 有效实现(3)为什么能正常运行 #

每个Flow的collect都运行在独立的Coroutine Scope中，对应独立的Job。独立Job之间无父子关系，取消其中一个不会影响另外两个。且DataStore初次读取速度远快于1秒，每个Flow都有足够时间完成第一次数据发射并赋值，最后依次取消各个Job即可停止Flow监听。

更优的实现方式 #

无需使用delay+取消Job这种繁琐且不可靠的方式，直接用Flow的first()操作符即可获取单次数据，它会在拿到第一个值后自动终止Flow，代码更简洁可靠：

串行读取版本 #

suspend fun setupDataStore(context: Context) {
    tempDataStore = TempDataStore(context)
    withContext(Dispatchers.IO) {
        die1.value = tempDataStore.getDieOne.first()!!.toInt()
        displayText.value = tempDataStore.getDisplayText.first()!!
        die2.value = tempDataStore.getDieTwo.first()!!.toInt()
    }
}

并发读取版本（效率更高） #

suspend fun setupDataStore(context: Context) {
    tempDataStore = TempDataStore(context)
    withContext(Dispatchers.IO) {
        val die1Deferred = async { tempDataStore.getDieOne.first()!!.toInt() }
        val displayTextDeferred = async { tempDataStore.getDisplayText.first()!! }
        val die2Deferred = async { tempDataStore.getDieTwo.first()!!.toInt() }
        
        die1.value = die1Deferred.await()
        displayText.value = displayTextDeferred.await()
        die2.value = die2Deferred.await()
    }
}

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