kotlin - 如何有效地跨多个协程实现速率限制？

Question

因此，假设我有一堆 couroutine 正在运行，它们与某些 Web 服务进行交互，并且由于我不想发送垃圾邮件，因此我想将请求限制为每 x 秒最多 1 个请求。为此，我可以使用一些这样的代码:

fun CoroutineScope.rateLimiter(tokens: SendChannel<Unit>, rate: Int) = launch {
    var lastToken = System.currentTimeMillis()
    while (isActive) {
        val currentTime = System.currentTimeMillis()
        if (currentTime - lastToken < rate) {
            delay(currentTime - lastToken)
        }
        tokens.send(Unit)
    }
}

fun CoroutineScope.request(tokens: ReceiveChannel<Unit>) = launch { 
    for (token in tokens) {
        //Do Web request
    }
}

1.) 这样做有效吗？

2.) 这不能扩展到将某些内容限制为 x 字节/秒或我需要从 Token Bucket 中请求 x 个 token 的内容，使用协程实现类似功能的最佳方法是什么？

Answer 1

如果您想跳过对可能产生的许可多于消耗的许可的工作和 channel 的依赖，然后在某个流程开始接受许可后就开始蜂拥而至，也许这就是您的解决方案。
(这里有一些 jvm 风格，但可替换为多平台)

import kotlin.math.max
import kotlinx.coroutines.delay
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock

class SimpleRateLimiter(eventsPerSecond: Double) {

    private val mutex = Mutex()

    @Volatile
    private var next: Long = Long.MIN_VALUE
    private val delayNanos: Long = (1_000_000_000L / eventsPerSecond).toLong()

    /**
     * Suspend the current coroutine until it's calculated time of exit
     * from the rate limiter
     */
    suspend fun acquire() {
        val now: Long = System.nanoTime()
        val until = mutex.withLock {
            max(next, now).also {
                next = it + delayNanos
            }
        }
        if (until != now) {
            delay((until - now) / 1_000_000)
        }
    }
}

它具有其他权衡。

当 nanoTime 接近 Long.MAX_VALUE 时的行为肯定已损坏。

maxDelay/timeout

没有实现

无法抓取多个白蚁

没有 tryAquire 实现

如果你想要一个 IntervalLimiter 允许 X 请求每 Y 秒，然后抛出异常，有 RateLimiter in Resilience4J
或者，如果您想要功能更全面的东西，我正在研究 PR to create both a RateLimiter and an IntervalLimiter in coroutines core project 。