scala - 带有 RateLimiter 的 ExecutionContext

Question

假设我有一个 HTTP 客户端来调用具有请求速率限制的服务器，例如1000 个请求/秒。我在 ExecutionContext 中实现了速率限制器像这样:
使用 RateLimiter 创建了一个有界阻塞队列 Guava

class MyBlockingQueue[A](capacity: Int, permitsPerSecond: Int) 
  extends ArrayBlockingQueue[A](capacity) {

  private val rateLimiter = RateLimiter.create(permitsPerSecond.toDouble)

  override def take(): A = {
    rateLimiter.acquire()
    super.take()
  }

  override def poll(timeout: Long, unit: TimeUnit): A = {
    rateLimiter.tryAcquire(timeout, unit) // todo: fix it
    super.poll(timeout, unit)
  }
}

创建了一个 ExecutionContext来自 ThreadPoolExecutor有了这个队列。

def createRateLimitingExecutionContext(numThreads: Int,
                                       capacity: Int,
                                       permitsPerSecond: Int): ExecutionContext = {
  val queue = new MyBlockingQueue[Runnable](capacity, permitsPerSecond)
  val executor = new ThreadPoolExecutor(numThreads, numThreads, 0L, TimeUnit.MILLISECONDS, queue)
  ExecutionContext.fromExecutor(executor)
}

现在我可以创建一个 ExecutionContext具有速率限制并将其传递给客户端:

implicit val ec = createRateLimitingThreadPoolExecutionContext(
  numThreads = 100,
  capacity = 1000,
  permitsPerSecond = 1000
)
httpGet("http://myserver.com/xyz") // create Futures with "ec"  

是否有意义 ？你会如何测试这个 ExecutionContext ?

Answer 1

看起来还不错，除了自定义执行上下文应该是显式的和/或在 httpGet 中管理或其封闭类，而不是全局隐式。
因为否则，当你写这样的东西时，例如:

    httpGet(foo)
     .recover("")
     .map(_.split(","))
     .map(_.map(_.toInt))
     .map(_.max)
     .foreach(println)

您最终消耗了 6(!) 个许可，而不是一个——也就是说，这相当于您提出了 6 个请求，这可能不是您想要的。