协程执行时间与回调执行时间

Question

我正在阅读“Functional Kotlin”一书，我刚刚从那本书中尝试了一些关于 Kotlin 并发编程的示例。我有以下并发代码的实现：

1. 带回调：

package com.freesoft.functional.coroutines

import kotlin.concurrent.thread

class CallbacksMain

class CallbackUserClient(private val client: UserClient) {
    fun getUser(id: Int, callback: (User) -> Unit) {
        thread {
            callback(client.getUser(id))
        }
    }
}

class CallbackFactClient(private val client: FactClient) {
    fun get(user: User, callback: (Fact) -> Unit) {
        thread {
            callback(client.getFact(user))
        }
    }
}

class CallBackUserRepository(private val userRepository: UserRepository) {
    fun getUserById(id: UserID, callback: (User?) -> Unit) {
        thread {
            callback(userRepository.getUserById(id))
        }
    }

    fun insertUser(user: User, callback: () -> Unit) {
        thread {
            userRepository.insertUser(user)
            callback()
        }
    }
}

class CallBackFactRepository(private val factRepository: FactRepository) {
    fun getFactByUserId(id: Int, callback: (Fact?) -> Unit) {
        thread {
            callback(factRepository.getFactByUserID(id))
        }
    }

    fun insertFact(fact: Fact, callback: () -> Unit) {
        thread {
            factRepository.insertFact(fact)
            callback()
        }
    }
}

class CallBackUserService(
    private val userClient: CallbackUserClient,
    private val factClient: CallbackFactClient,
    private val userRepository: CallBackUserRepository,
    private val factRepository: CallBackFactRepository
) : UserService {
    override fun getFact(id: UserID): Fact {
        var aux: Fact? = null
        userRepository.getUserById(id) { user ->
            if (user == null) {
                userClient.getUser(id) { userFromClient ->
                    userRepository.insertUser(userFromClient) {}
                    factClient.get(userFromClient) { fact ->
                        factRepository.insertFact(fact) {}
                        aux = fact
                    }
                }
            } else {
                factRepository.getFactByUserId(id) { fact ->
                    if (fact == null) {
                        factClient.get(user) { factFromClient ->
                            factRepository.insertFact(factFromClient) {}
                            aux = factFromClient
                        }
                    } else {
                        aux = fact
                    }
                }
            }
        }
        while (aux == null) {
            Thread.sleep(2)
        }
        return aux!!
    }
}


fun main(args: Array<String>) {
    fun execute(userService: UserService, id: Int) {
        val (fact, time) = inTime {
            userService.getFact(id)
        }
        println("fact = $fact")
        println("time = $time ms.")
    }

    val userClient = MockUserClient()
    val callbackUserClient = CallbackUserClient(userClient)
    val factClient = MockFactClient()
    val callBackFactClient = CallbackFactClient(factClient)
    val userRepository = MockUserRepository()
    val callbackUserRepository = CallBackUserRepository(userRepository)
    val factRepository = MockFactRepository()
    val callBackFactRepository = CallBackFactRepository(factRepository)

    val callBackUserService = CallBackUserService(
        userClient = callbackUserClient,
        factClient = callBackFactClient,
        userRepository = callbackUserRepository,
        factRepository = callBackFactRepository
    )

    execute(callBackUserService, 1)
    execute(callBackUserService, 2)
    execute(callBackUserService, 1)
    execute(callBackUserService, 2)
    execute(callBackUserService, 3)
    execute(callBackUserService, 4)
    execute(callBackUserService, 5)
    execute(callBackUserService, 10)
    execute(callBackUserService, 100)
}

和 2. 使用协程：

package com.freesoft.functional.coroutines

import kotlinx.coroutines.*

class CoroutinesUserService(
    private val userClient: UserClient,
    private val factClient: FactClient,
    private val userRepository: UserRepository,
    private val factRepository: FactRepository
) : UserService {
    override fun getFact(id: UserID): Fact = runBlocking {
        val user = async { userRepository.getUserById(id) }.await()
        if (user == null) {
            val userFromService = async { userClient.getUser(id) }.await()
            launch { userRepository.insertUser(userFromService) }
            getFact(userFromService)
        } else {
            async {
                factRepository.getFactByUserID(id) ?: getFact(user)
            }.await()
        }
    }

    private suspend fun getFact(user: User): Fact {
        val fact: Deferred<Fact> = withContext(Dispatchers.Default) {
            async { factClient.getFact(user) }
        }
        coroutineScope {
            launch { factRepository.insertFact(fact.await()) }
        }
        return fact.await()

    }
}

fun main(args: Array<String>) {
    fun execute(userService: UserService, id: Int) {
        val (fact, time) = inTime {
            userService.getFact(id)
        }
        println("fact = $fact")
        println("time = $time ms.")
    }

    val userClient = MockUserClient()
    val factClient = MockFactClient()
    val userRepository = MockUserRepository()
    val factRepository = MockFactRepository()

    val coroutinesUserService = CoroutinesUserService(userClient, factClient, userRepository, factRepository)

    execute(coroutinesUserService, 1)
    execute(coroutinesUserService, 2)
    execute(coroutinesUserService, 1)
    execute(coroutinesUserService, 2)
    execute(coroutinesUserService, 3)
    execute(coroutinesUserService, 4)
    execute(coroutinesUserService, 5)
    execute(coroutinesUserService, 10)
    execute(coroutinesUserService, 100)
}

Here are the mocks that I'm using `UserClient,FactClient,UserRepository and FactRepository`:

class MockUserClient : UserClient {
    override fun getUser(id: UserID): User {
        println("MockUserClient.getUser")
        Thread.sleep(500)
        return User(id, "Foo", "Bar", Gender.FEMALE)
    }
}

class MockFactClient : FactClient {
    override fun getFact(user: User): Fact {
        println("MockFactClient.getFact")
        Thread.sleep(500)
        return Fact(Random().nextInt(), "FACT ${user.firstName}, ${user.lastName}", user)
    }
}

class MockUserRepository : UserRepository {
    private val users = hashMapOf<UserID, User>()

    override fun getUserById(id: UserID): User? {
        println("MockUserRepository.getUserById")
        Thread.sleep(200)
        return users[id]
    }

    override fun insertUser(user: User) {
        println("MockUserRepository.insertUser")
        Thread.sleep(200)
        users[user.id] = user
    }
}

class MockFactRepository : FactRepository {
    private val facts = hashMapOf<UserID, Fact>()

    override fun getFactByUserID(userID: UserID): Fact? {
        println("MockFactRepository.getFactByUserId")
        Thread.sleep(200)
        return facts[userID]
    }

    override fun insertFact(fact: Fact) {
        println("MockFactRepository.insertFact")
        Thread.sleep(200)
        facts[fact.user?.id ?: 0] = fact
    }
}

我的问题是：即使对模拟对象的请求总和应该在 1.2 秒左右，为什么使用协程实现的时间成本更高？在回调实现中，我收到了正确的执行时间（1.2 秒），但在协程实现中，我收到了大约 1.4 秒的执行时间。

Answer 1

这里有两个误解：

1. 协程比线程快
2. 你实际上是在有效地使用协程

让我们从第一个开始。
协程并不比线程快。不过，它们更并发。这意味着如果你有数千个协程，你仍然可以。如果您有数千个线程 - 您将耗尽内存。

但是您的测试根本不具有侵略性：

execute(callBackUserService, 1)
execute(callBackUserService, 2)
execute(callBackUserService, 1)
execute(callBackUserService, 2)
execute(callBackUserService, 3)
execute(callBackUserService, 4)
execute(callBackUserService, 5)
execute(callBackUserService, 10)
execute(callBackUserService, 100)

即使您为每个 启动 6 个线程execute，也不会产生足够的开销。做类似的事情

 repeat(100_000) {
     execute(callBackUserService, 100)
 }

你会看到不同之处。

不说第二点了。

override fun getFact(id: UserID): Fact = runBlocking {
    val user = async { userRepository.getUserById(id) }.await()
    ...
}

什么runBlocking是阻止您的调用执行上下文。由于您在main线程上，因此您只在一个线程上运行。通过使用async {}.await()相同的方式，您不会从任何并发中受益，因为您在线程上运行您的任务，然后立即阻止它。

一旦你解决了这两个并发问题，还有第三个问题等着你：Thread.sleep(200)在你的模拟中。

您可能期望的是每个协程将被阻塞 200 毫秒。实际上会发生的是所有协程都被阻塞了 200 毫秒，因为它们共享相同的执行池。如果你想用协程测试你的模拟，你必须使用delay()，否则，你没有测试任何东西。