并发修改如何与协程一起工作？

Question

我正在阅读这个coroutines 动手教程Coroutines-Channels。所以有这个任务是使用channels同时获取贡献者并显示中间进度，参见Here

以下是建议解决方案的sn-p

suspend fun loadContributorsChannels(
service: GitHubService,
req: RequestData,
updateResults: suspend (List<User>, completed: Boolean) -> Unit) = coroutineScope {

........
.........
val channel = Channel<List<User>>()
for (repo in repos) {
    launch {
        val users = service.getRepoContributors(req.org, repo.name) // suspend function
            .also { logUsers(repo, it) }
            .bodyList()
        channel.send(users) // suspend function
    }
}
var allUsers = emptyList<User>()
repeat(repos.size) {
    val users = channel.receive()  // suspend function
    allUsers = (allUsers + users).aggregate()
    updateResults(allUsers, it == repos.lastIndex) // suspend function
}
}

函数loadContributorsChannels() 在使用Default dispatcher.See here 的coroutine 内调用。我有 2 个问题。

1. 在上面的 sn-p 中，allUsers 被同时修改，因为我们已经在使用 Default dispatcher 的 coroutine 中？

2. 如果我像下面这样更改代码序列，为什么会得到不正确的结果？上面的代码和下面的sn-p有什么区别？

   val contributorsChannel = Channel<List<User>>()
   var contributors = emptyList<User>()
   
   for(repo in repos) { 
       launch {
           val contributorsPerRepo = service
               .getRepoContributors(req.org, repo.name) // suspend function
               .also { logUsers(repo, it) }
               .bodyList()
   
           contributors = (contributors + contributorsPerRepo).aggregate()
           contributorsChannel.send(contributors)  // suspend function
       }
   }
   
   repeat(repos.size) {
       updateResults(contributorsChannel.receive(), it == repos.lastIndex) // suspend functions
   }
   

这是因为并发修改还是我遗漏了什么？

Answer 1

在原始代码中，顶层协程是唯一使用allUsers的协程。这是它的本地状态。

在您的代码中，contributors 是一个由所有协程共享并同时更新的变量。

原始代码正确地将 Channel 作为一种同步机制，将所有并发计算扇入到一个协程中，该协程收集结果并使用它们。