scala – Akka Http – 主机级客户端API Source.queue模式

我们开始实现docs： http://doc.akka.io/docs/akka-http/current/scala/http/client-side/host-level.html#examples中提到的Source.queue [HttpRequest]模式

这是文档中的(简化)示例

val poolClientFlow = Http()
  .cachedHostConnectionPool[Promise[HttpResponse]]("akka.io")

val queue =
  Source.queue[(HttpRequest,Promise[HttpResponse])](
     QueueSize,OverflowStrategy.dropNew
  )
    .via(poolClientFlow)
    .toMat(Sink.foreach({
      case ((Success(resp),p)) => p.success(resp)
      case ((Failure(e),p))    => p.failure(e)
    }))(Keep.left)
    .run()

def queueRequest(request: HttpRequest): Future[HttpResponse] = {
  val responsePromise = Promise[HttpResponse]()
  queue.offer(request -> responsePromise).flatMap {
    case QueueOfferResult.Enqueued    => responsePromise.future
    case QueueOfferResult.Dropped     => Future.failed(new RuntimeException("Queue overflowed. Try again later."))
    case QueueOfferResult.Failure(ex) => Future.failed(ex)
    case QueueOfferResult.QueueClosed => Future.failed(new RuntimeException("Queue was closed (pool shut down) while running the request. Try again later."))
  }
}

val responseFuture: Future[HttpResponse] = queueRequest(HttpRequest(uri = "/"))

文档声明使用Source.single(request)是一个反模式,应该避免.但是,它没有说明使用Source.queue的原因和含义.

At this place we previously showed an example that used the Source.single(request).via(pool).runWith(Sink.head).
In fact,this is an anti-pattern that doesn’t perform well. Please either supply requests using a queue or in a streamed fashion as shown below.

Source.queue的优点

>流程只实现一次(可能是性能提升？).但是,如果我正确理解了akka-http实现,那么每个连接都会实现一个新流程,所以这似乎不是一个很大的问题
>使用OverflowStrategy进行显式背压处理并匹配QueueOfferResult

Source.queue的问题

当我们在应用程序中开始实现此模式时,会出现这些问题.

Source.queue不是线程安全的

The queue implementation is not thread safe.当我们在不同的路线/演员中使用队列时,我们有这样的场景：

排队的请求可以覆盖最新的排队请求,从而导致未解决的Future.

UPDATE

此问题已在akka/akka/issues/23081中解决.队列实际上是线程安全的.

过滤？

过滤请求时会发生什么？例如.当有人改变实施时

Source.queue[(HttpRequest,Promise[HttpResponse])](
    QueueSize,OverflowStrategy.dropNew)
  .via(poolClientFlow)
  // only successful responses
  .filter(_._1.isSuccess)
  // failed won't arrive here
  .to(Sink.foreach({
    case ((Success(resp),p)) => p.success(resp)
    case ((Failure(e),p)) => p.failure(e)
  }))

未来会不会解决？使用单个请求流程,这很简单：

Source.single(request).via(poolClientFlow).runWith(Sink.headOption)

QueueSize vs max-open-request？

QueueSize和max-open-requests之间的区别尚不清楚.最后,两者都是缓冲区.我们的实现最终使用QueueSize == max-open-requests

Source.single()的缺点是什么？

到目前为止,我发现在Source.single上使用Source.queue有两个原因

>性能 – 仅实现流程一次.但是根据this answer,它应该不是问题
>明确配置背压并处理故障情况.在我看来,ConnectionPool具有足够的负载处理能力.可以映射结果的未来并处理异常.

提前致谢,
缪奇

val queue = ??? //as in the example in your question

queue.offer(httpRequest1)
queue.offer(httpRequest2)
queue.offer(httpRequest3)

val allRequests = Iterable(httpRequest1,httpRequest2,httpRequest3)

//no queue necessary
val allResponses : Future[Seq[HttpResponse]] = 
  Source(allRequests)
    .via(poolClientFlow)
    .to(Sink.seq[HttpResponse])
    .run()

import scala.io.{Source => ioSource}

val consoleLines : () => Iterator[String] = 
  () => ioSource.stdin.getLines()

Source
  .fromIterator(consoleLines)
  .map(consoleLine => HttpRequest(GET,uri = Uri(consoleLine)))
  .via(poolClientFlow)
  .to(Sink.foreach[HttpResponse](println))
  .run()