带有改造的协程线程安全

Posted 2023-02-16

【中文标题】带有改造的协程线程安全

【英文标题】：Coroutine Thread Safety with Retrofit

【发布时间】：2021-12-07 22:51:45

【问题描述】：

关于使用协程启动网络请求的线程安全性的所有信息，我仍然有点麻烦。

假设我们有以下用例，我们获得了一个用户列表，对于每个用户，我将做一些特定的检查，这些检查必须通过对 API 的网络请求运行，给我一些信息关于这个用户。

userCheck 发生在库中，该库不公开挂起函数，但仍使用回调。 在这个库中，我见过这样的代码来启动每个网络请求：

internal suspend fun <T> doNetworkRequest(request: suspend () -> Response<T>): NetworkResult<T> 
    return withContext(Dispatchers.IO) 
        try 
            val response = request.invoke()
            ...

根据文档，Dispatchers.IO 可以使用多个线程来执行代码，请求函数也只是来自 Retrofit API 的函数。

所以我所做的是为每个用户启动请求，并使用单个 resultHandler 对象，它将结果添加到列表中并检查结果列表的长度是否等于用户列表的长度，如果是, 然后所有 userChecks 都完成了，我知道我可以对结果做一些事情，这些结果需要一起返回。

val userList: List<String>? = getUsers()
val userCheckResultList = mutableListOf<UserCheckResult>()
val handler = object : UserCheckResultHandler 
                  override fun onResult(
                        userCheckResult: UserCheckResult?
                  ) 
                        userCheckResult?.let 
                            userCheckResultList.add(
                                it
                            )
                        
                        if (userCheckResultList.size == userList?.size) 
                            doSomethingWithResultList()
                            print("SUCCESS")
                        
                    
                

userList?.forEach 
    checkUser(it, handler)

我的问题是：这个实现是线程安全的吗？据我所知，Kotlin 对象应该是线程安全的，但我得到反馈说这可能不是最好的实现：D

但理论上，即使请求同时异步启动多个，一次也只有一个可以访问结果处理程序正在运行的线程的锁，并且不会出现竞争条件或添加问题列表中的项目并比较大小。

我错了吗？ 有什么方法可以更好地处理这种情况？

【参考方案1】：

如果您正在并行执行多个请求 - 它不是。 List 不是线程安全的。但这很简单。创建一个Mutex 对象，然后将您的操作锁定在列表中，如下所示：

val lock = Mutex()
val userList: List<String>? = getUsers()
val userCheckResultList = mutableListOf<UserCheckResult>()
val handler = object : UserCheckResultHandler 
                  override fun onResult(
                        userCheckResult: UserCheckResult?
                  ) 
                        lock.withLock 
                            userCheckResult?.let 
                                userCheckResultList.add(
                                    it
                                )
                            
                            if (userCheckResultList.size == userList?.size) 
                                doSomethingWithResultList()
                                print("SUCCESS")
                            
                        
                    
                

userList?.forEach 
    checkUser(it, handler)

我必须补充一点，整个解决方案看起来很老套。我会完全走其他路线。运行所有请求，将这些请求包装在 async // network request 中，这将返回 Deferred 对象。将此对象添加到某个列表中。之后使用awaitAll() 等待所有这些延迟对象。像这样：

val jobs = mutableListOf<Job>()
userList?.forEach 
   // i assume checkUser is suspendable here
   jobs += async  checkUser(it, handler) 


// wait for all requests
jobs.awaitAll()

// After that you can access all results like this:
val resultOfJob0 = jobs[0].getCompleted()

【讨论】：

首先，感谢您的回答！提议的解决方案看起来确实更好，但是 checkUser 可暂停的假设是不正确的，这让我选择了另一种 hacky 方式。这个函数在底层确实使用了协程，但由于与 Java 的互操作性，它没有公开挂起函数。

@Igor 哦，好的，所以带锁的版本可能是你的解决方案:)

我还不清楚一件事：如果 Kotlin 对象是线程安全的，即使我同时启动多个请求，一次也应该只有一个能够触发回调，或者我是错误的？如果当时只有一个触发回调和 onResult 函数中的代码，为什么它会使向列表中添加对象不是线程安全的？ - 或者简而言之：如果列表不是从协程本身写入，而是从线程安全对象写入，我怎么还需要锁？

@Igor 我不确定你从哪里得到“Kotlin 对象是线程安全的”的信息。默认情况下，没有对象是线程安全的。由于请求在 IO 调度程序中运行，因此您可以并且可能会并行调用回调。 kotlin 中有一些内置类是线程安全的，但您必须逐个评估它们

我的意思是最终变量的匿名对象初始化称为处理程序。根据这个线程：***.com/questions/30179793/… 对象，它们是单例，应该是线程安全的，还是我解释这个答案错了​​？