使用 Akka Stream 流式传输巨大的 json答案

【问题标题】：Streaming huge json with Akka Stream使用 Akka Stream 流式传输巨大的 json
【发布时间】：2017-11-12 12:04:24
【问题描述】：

我有一个 json 平板的巨大 http 响应问题，其中只有一部分是兴趣点。我无法更改响应结构。这是一个例子

{
  "searchString": "search",
  "redirectUrl": "",
  "0": {
    "numRecords": 123,
    "refinementViewModelCollector": {},
//    Lots of data here
    "results": [
      {
        "productCode": "123",
        "productShortDescription": "Desc",
        "brand": "Brand",
        "productReview": {
          "reviewScore": 0
        },
        "priceView": {
          "salePriceDisplayable": false,
        },
        "productImageUrl": "url",
        "alternateImageUrls": [
          "url1"
        ],
        "largeProductImageUrl": "url4",
        "videoUrl": ""
      },
      {
        "productCode": "124",
        "productShortDescription": "Desc",
        "brand": "Brand",
        "productReview": {
          "reviewScore": 0
         },
        "priceView": {
          "salePriceDisplayable": false,
        },
        "preOrder": false,
        "productImageUrl": "url",
        "alternateImageUrls": [
          "url1"
        ],
        "largeProductImageUrl": "url4",
        "videoUrl": ""
      }
    ]
    //lots of data here
  }
}

我的兴趣点是 results Jason 数组中的条目，但它们位于 json 中间

我像这样创建了一个小型 Play WS 客户端：

val wsClient: WSClient = ???
val ret = wsClient.url("url").stream()
ret.flatMap { response =>
  response.body.via(JsonFraming.objectScanner(1024))
    .map(_.utf8String)
    .runWith(Sink.foreach(println))
}

这将不起作用，因为它将整个 json 平板作为 Json 对象。我需要跳过一些数据，直到"results": 条目出现在流中，然后开始解析条目并跳过所有其余部分。任何想法如何做到这一点？

【问题讨论】：

使用jsoniter-scala FTW！它可以立即将您的字节数组 java.io.InputStream 或 java.nio.ByteBuffer 解析为您的数据结构，从而非常有效地跳过不需要的 JSON 值。请检查 ExtractFields 基准测试的here 结果，以查看每个线程每秒可能超过 1GB 的吞吐量

标签： json scala playframework akka akka-stream

【解决方案1】：

查看Alpakka's JSON module，它可以流式传输嵌套 JSON 结构的特定部分：

response
  .body
  .via(JsonReader.select("$.0.results[*]"))
  .map(_.utf8String)
  .runWith(Sink.foreach(println)) // or runForeach(println)

【讨论】：

【解决方案2】：

有支持解析为流的解析器。对于一个很好的例子，看看这个 Circe 例子https://github.com/circe/circe/tree/master/examples/sf-city-lots

【讨论】：

所描述的方法适用于流式 JSON 或大型顶级数组，但不能解决对象中包含大型数组的情况。

【解决方案3】：

我希望对这个问题有一个更好的、特定于 Scala 的答案，但请查看 Google 的 GSON 库文档中的“混合读取示例”：

https://sites.google.com/site/gson/streaming

Gson 还支持混合流和对象模型访问。这让您的应用程序拥有两全其美：对象模型访问的生产力和流式传输的效率 ... 此代码读取包含消息数组的 JSON 文档。它以流的形式逐步遍历数组元素，以避免将完整的文档加载到内存中。它很简洁，因为它使用 Gson 的对象模型来解析单个消息

这应该具有出色的内存性能（代码从 Java InputStream 中读取，因此完整的结构永远不会在内存中），但可能需要一些努力才能将结果放入 Scala 案例类中。

【讨论】：