akka-stream+akka-http生命周期

Question

// example of stream per request

val connectionFlow = Http().cachedHostConnectionPool[UUID](host, port)
val httpFlow: Flow[HttpRequest, Try[HttpResponse], NotUsed] =
    Flow[HttpRequest]
      .map(req => req -> UUID.randomUUID()) // I don't care about id because it's a single request per stream.
      .via(connectionFlow)
      .map { case (response, _) => response }

val result = Range(1 to 5).foreach{ i => {
  Source.single(i)
    .map(HttpRequest(...))
    .via(httpFlow)
    .mapAsync(1) {
       // response handling logic
    }
    .runWith(Sink.last)
})


// example of stream per request with long running http stream

// as defined in http://doc.akka.io/docs/akka-http/current/scala/http/client-side/host-level.html#using-the-host-level-api-with-a-queue
def queueRequest(request: HttpRequest): Future[HttpResponse]

val result = Range(1 to 5).foreach{ i => {
  Source.single(i)
    .map(HttpRequest(...))
    .mapAsync(1)(queueRequest)
    .mapAsync(1) {
       // somehow reconcile request with response?
       // response handling logic
    }
    .runWith(Sink.last)
})

我确实尝试实现了这两种解决方案，但在实现的每个阶段都有许多设计选择，因此即使在一条“正确”的道路上，似乎也很容易搞砸。

¹虽然我相信它是可以忽略不计的，而且是akka-http服务器运行的相同方式。

Answer 1

通常，使用单个连接flow并通过该单个流分派所有请求要好得多。主要原因是，新的物化可能会导致每次形成新的连接（这取决于您的连接池设置）。

您认为这会导致一些并发症,这是正确的：

排序：通过提供一个随机的uuid作为传递给连接流的元组中的第二个值，您将消除将请求与响应相关联的能力。元组中额外的t值可以用作“相关ID”，以了解您从流中获得的httpresponse。在您的特定示例中，您可以使用创建的范围中的初始int:

val responseSource : Source[(Try[HttpResponse], Int), _] = 
  Source
    .fromIterator( () => Iterator range (0,5) )
    .map(i => HttpRequest(...) -> i)
    .via(connectionFlow)