Akka HTTP 源流与常规请求处理

Question

使用Source Streaming 与处理请求的常规方式相比有什么优势？我的理解是在这两种情况下

1. TCP 连接将被重用
2. 将在客户端和服务器之间施加背压

我可以看到源流的唯一优势是如果响应非常大并且客户端更喜欢以较小的块消耗它。

我的用例是我有一个很长的用户列表（数百万），我需要调用一个对用户执行一些过滤并返回一个子集的服务。

目前，在服务器端我公开了一个批处理 API，在客户端，我只是将用户分成 1000 个块，并使用 Akka HTTP Host API 并行进行 X 个批处理调用。

我正在考虑切换到 HTTP 流式传输，但无法完全弄清楚价值是什么

Answer 1

您还缺少另一个巨大的好处：内存效率。通过拥有流式管道、客户端/服务器/客户端，各方可以安全地处理数据，而不会冒着破坏内存分配的风险。这在服务器端特别有用，您总是必须假设客户端可能会做一些恶意的事情......

创建客户端请求

假设您的数百万用户的最终来源是一个文件。您可以从此文件创建流源：

val userFilePath : java.nio.file.Path = ???

val userFileSource = akka.stream.scaladsl.FileIO(userFilePath)

您可以使用此源创建将用户流式传输到服务的 http 请求：

import akka.http.scaladsl.model.HttpEntity.{Chunked, ChunkStreamPart}
import akka.http.scaladsl.model.{RequestEntity, ContentTypes, HttpRequest}

val httpRequest : HttpRequest = 
  HttpRequest(uri = "http://filterService.io", 
              entity = Chunked.fromData(ContentTypes.`text/plain(UTF-8)`, userFileSource))

此请求现在会将用户流式传输到服务，而不会将整个文件消耗到内存中。一次只会缓冲数据块，因此，您可以发送可能无限的请求用户数量和您的客户都可以。

服务器请求处理

同样，您的服务器可以设计为接受具有可能无限长的实体的请求。

你的问题是说服务会过滤用户，假设我们有过滤功能：

val isValidUser : (String) => Boolean = ???

这可用于过滤传入的请求实体并创建一个响应实体来提供响应：

import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.model.HttpResponse
import akka.http.scaladsl.model.HttpEntity.Chunked

val route = extractDataBytes { userSource =>
  val responseSource : Source[ByteString, _] = 
    userSource
      .map(_.utf8String)
      .filter(isValidUser)
      .map(ByteString.apply)

  complete(HttpResponse(entity=Chunked.fromData(ContentTypes.`text/plain(UTF-8)`, 
                                                responseSource)))
}

客户端响应处理

客户端可以类似地处理过滤后的用户，而无需将它们全部读入内存。例如，我们可以分派请求并将所有有效用户发送到控制台：

import akka.http.scaladsl.Http

Http()
  .singleRequest(httpRequest)
  .map { response =>
    response
      .entity
      .dataBytes
      .map(_.utf8String)
      .foreach(System.out.println)
  }