kotlin - 带有请求队列的 Kotlin 服务

Question

我想用以下 API 设计一个服务:

suspend fun getUsers(request: Request): List<User>

在幕后我会向服务器发送一个请求(不管如何，但可以说它是一个响应式(Reactive) WebClient )，但这里有一个技巧:我只能每 500 毫秒发送一次请求，否则我会得到一个错误。

有人可以推荐我如何实现它，当我调用 getUsers 时从它挂起的协程中，工作单元被添加到具有此方法的服务的某个队列中，然后在某个时间点执行并返回结果？

我假设我可以使用一些 ReceiveChannel作为一个队列，有一个 for使用 delay 循环其元素在里面，但我有点迷失了把这个逻辑放在哪里。这应该像一个永远运行并被 getUsers 调用的后台方法吗？ ？可能是 close方法永远不会被调用，所以这个方法也可以被挂起，但是我如何将值从这个无限运行的方法传回给getUsers需要结果吗？

编辑

目前我正在考虑这样的解决方案:

private const val REQUEST_INTERVAL = 500

@Service
class DelayedRequestSenderImpl<T> : DelayedRequestSender<T> {
    private var lastRequestTime: LocalDateTime = LocalDateTime.now()
    private val requestChannel: Channel<Deferred<T>> = Channel()

    override suspend fun requestAsync(block: () -> T): Deferred<T> {
        val deferred = GlobalScope.async(start = CoroutineStart.LAZY) { block() }
        requestChannel.send(deferred)
        return deferred
    }

    @PostConstruct
    private fun startRequestProcessing() = GlobalScope.launch {
        for (request in requestChannel) {
            val now = LocalDateTime.now()
            val diff = ChronoUnit.MILLIS.between(lastRequestTime, now)
            if (diff < REQUEST_INTERVAL) {
                delay(REQUEST_INTERVAL - diff)
                lastRequestTime = now
            }
            request.start()
        }
    }
}

我在这里看到的问题是我必须生成类来生成 requestChannel通用的，因为请求的结果可能是任何东西。但这意味着 DelayedRequestSender 的每个实例将绑定(bind)到特定类型。关于如何避免这种情况的任何建议？

编辑 2

这是一个改进的版本。我目前看到的唯一可能的流程是我们必须制作 @PostConstruct公共(public)方法，以便在我们需要或使用反射时编写任何测试。

这个想法是不使用 GlobalScope还有一个单独的Job对于处理方法。这是一个很好的方法吗？

interface DelayingSupplier {
    suspend fun <T> supply(block: () -> T): T
}

@Service
class DelayingSupplierImpl(@Value("\${vk.request.interval}") private val interval: Int) : DelayingSupplier {
    private var lastRequestTime: LocalDateTime = LocalDateTime.now()
    private val requestChannel: Channel<Deferred<*>> = Channel()
    private val coroutineScope = CoroutineScope(EmptyCoroutineContext)

    override suspend fun <T> supply(block: () -> T): T {
        val deferred = coroutineScope.async(start = CoroutineStart.LAZY) { block() }
        requestChannel.send(deferred)
        return deferred.await()
    }

    @PostConstruct
    fun startProcessing() = coroutineScope.launch(context = Job(coroutineScope.coroutineContext[Job])) {
        for (request in requestChannel) {
            val now = LocalDateTime.now()
            val diff = ChronoUnit.MILLIS.between(lastRequestTime, now)
            if (diff < interval) {
                delay(interval - diff)
            }
            lastRequestTime = LocalDateTime.now()
            request.start()
        }
    }
}

Answer 1

我会推荐:

• 将泛型下推到函数级别
• 使用 actor 而不是你的协程实现(但你可能更喜欢这个)。

无论哪种方式，此解决方案都应允许您使用队列的单个实例来处理所有请求的延迟，而不管返回类型如何。 (抱歉，我重命名了一些东西以帮助我自己的概念化，希望这仍然有意义):

private const val REQUEST_INTERVAL = 500

interface DelayedRequestHandler {

    suspend fun <T> handleWithDelay(block: () -> T): T

}

class DelayedRequestHandlerImpl(requestInterval: Int = REQUEST_INTERVAL) : DelayedRequestHandler, CoroutineScope {
    private val job = Job()
    override val coroutineContext = Dispatchers.Unconfined + job
    private val delayedHandlerActor = delayedRequestHandlerActor(requestInterval)

    override suspend fun <T> handleWithDelay(block: () -> T): T {
        val result = CompletableDeferred<T>()
        delayedHandlerActor.send(DelayedHandlerMsg(result, block))
        return result.await()
    }
}

private data class DelayedHandlerMsg<RESULT>(val result: CompletableDeferred<RESULT>, val block: () -> RESULT)

private fun CoroutineScope.delayedRequestHandlerActor(requestInterval: Int) = actor<DelayedHandlerMsg<*>>() {
    var lastRequestTime: LocalDateTime = LocalDateTime.now()
    for (message in channel) {
        try {
            println("got a message processing")
            val now = LocalDateTime.now()
            val diff = ChronoUnit.MILLIS.between(lastRequestTime, now)
            if (diff < requestInterval) {
                delay(requestInterval - diff)
            }
            lastRequestTime = LocalDateTime.now()
            @Suppress("UNCHECKED_CAST")
            val msgCast = message as DelayedHandlerMsg<Any?>
            val result = msgCast.block()
            println(result)
            msgCast.result.complete(result)
        } catch (e: Exception) {
            message.result.completeExceptionally(e)
        }
    }
}


fun main() = runBlocking {
    val mydelayHandler = DelayedRequestHandlerImpl(2000)
    val jobs = List(10) {
        launch {
            mydelayHandler.handleWithDelay {
                "Result $it"
            }
        }
    }
    jobs.forEach { it.join() }
}