使用Akka Http转换流畅的流数据并发送分块响应

您可以使用 mapAsync :

val src =
  Source.fromPublisher(db.stream(getRds))
        .mapAsync(parallelism = 3)(compute)

complete(src)

根据需要调整平行度级别。

请注意,您可能需要配置中提到的一些设置 Slick documentation :

注意:一些数据库系统可能需要以某种方式设置会话参数,以支持流式传输,而无需在客户端的内存中一次性缓存所有数据。例如,PostgreSQL要求 .withStatementParameters(rsType = ResultSetType.ForwardOnly, rsConcurrency = ResultSetConcurrency.ReadOnly, fetchSize = n) (具有所需的页面大小 n )以及 .transactionally 用于正确的流式处理。

例如,如果您使用的是PostgreSQL,那么 Source 可能如下所示:

val src =
  Source.fromPublisher(
    db.stream(
      getRds.withStatementParameters(
        rsType = ResultSetType.ForwardOnly,
        rsConcurrency = ResultSetConcurrency.ReadOnly,
        fetchSize = 10
      ).transactionally
    )
  ).mapAsync(parallelism = 3)(compute)

您需要有一种方法来marshall TweetNew,并且如果您发送长度为0的区块,客户端可能会关闭连接。

此代码适用于curl:

case class TweetNew(str: String)

def compute(string: String) : Future[TweetNew] = Future {
  TweetNew(string)
}

val route = path("hello") {
  get {
    val byteString: Source[ByteString, NotUsed] = Source.apply(List("t1", "t2", "t3"))
      .mapAsync(2)(compute)
      .map(tweet => ByteString(tweet.str + "\n"))
    complete(HttpEntity(ContentTypes.`text/plain(UTF-8)`, byteString))
  }
}