如何在 Akka Stream 2 中处理来自 WSClient 的流数据答案

【问题标题】：How to handle streaming data that comes from WSClient in Akka Stream 2如何在 Akka Stream 2 中处理来自 WSClient 的流数据
【发布时间】：2017-03-04 07:00:42
【问题描述】：

如何在下面的代码中返回concurrent.Future：

val nextMeetup = ws.url(url).stream()
        .flatMap( response => response.body
          .via(framing)
          .map(_.utf8String)
          .map(_ + "\n")
        )

类型不匹配错误：找到：akka.stream.scaladsl.Source[String,_$2] 必需：scala.concurrent.Future[?]

我的 url 拆分 json 数据流 - 它不是 twitter 流

请解释解决这个问题。

【问题讨论】：

标签： scala playframework akka-stream

【解决方案1】：

目前还不清楚您想从响应中得到什么。 stream() 返回一个 Future，因此如果你在它上面 flatMap，你还需要从处理程序返回一个未来。你到底想在那里返回什么？例如，如果您想获取带有整个响应正文的Future[String]，您可以使用runReduce(_ + _)：

val result: Future[String] = ws.url(url).stream()
    .flatMap(response => response.body
      .via(framing)
      .map(_.utf8String)
      .map(_ + "\n")
      .runReduce(_ + _)
    )

runReduce(f: (U, U) => U) 返回一个Future[U]，也就是说，在你的情况下它是Future[String]。如果你想通过其他函数分别处理传入流的每个元素，你可以使用runForeach：

ws.url(url).stream()
    .flatMap( response => response.body
      .via(framing)
      .map(_.utf8String)
      .map(_ + "\n")
      .runForeach(s => handleString(s))
    )

如果没有关于您想要做什么的更多细节，很难提供更具体的答案。

更新：如果你想限制来自外部服务器的消息，你可以使用内置的throttle 组合器：

val result: Future[Source[String, _]] = ws.url(url)
  .stream()
  .map { response =>
    response.body
      .via(framing)
      .map(_.utf8String)
      .map(_ + "\n")
      .throttle(10, 1.second, 10, ThrottleMode.Shaping)
  }

这里，result 未来将包含Strings 的流，当具体化时，每秒最多会产生 10 个元素，必要时会反压。您可以在我上面链接的文档中找到更多信息。

【讨论】：

谢谢 - 输出是一个 json 字符串。我想获取字符串并在 scala 中对其进行表述并更新我的案例类。
@SakarSR，我还是不明白，抱歉。您所说的“措辞”是什么意思？考虑写一个你想在没有反应流的情况下做什么的例子。
用户名：@Vladimir url 网络服务器不间断地发出 json 数据。我想捕获数据并对数据进行一些计算并将其传递给前端。我也想减慢我的网络服务器（回压）它。计划使用 Akka 流。我希望它清楚。
那么我是否正确理解您希望以流方式将数据从外部 Web 服务传递到您自己的 Web 服务的客户端？也就是比如你的服务的客户端是通过websocket连接到它的？
@Vladimr 是的，这正是我想要的。我通过 websocket 使用了流式计时器，它工作正常。但在这种情况下，我对外部 Web 服务没有任何控制权，一旦我调用外部 Web 服务，它就会开始推送一些 json 数据。我想控制它。