Akka http-client无法使用来自服务器的所有数据流

Question

我正在尝试编写一个简单的akka​​流休息端点和客户端来使用此流。但后来我尝试运行服务器和客户端，客户端只能消耗部分流。我在执行期间看不到任何异常。

这是我的服务器和客户端：

import akka.NotUsed
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.common.{EntityStreamingSupport, JsonEntityStreamingSupport}
import akka.http.scaladsl.server.Directives._
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import akka.stream.{ActorAttributes, ActorMaterializer, Attributes, Supervision}
import akka.stream.scaladsl.{Flow, Source}
import akka.util.ByteString
import spray.json.DefaultJsonProtocol

import scala.io.StdIn
import scala.util.Random

object WebServer {

  object Model {
    case class Person(id: Int = Random.nextInt(), fName: String = Random.nextString(10), sName: String = Random.nextString(10))
  }

  object JsonProtocol extends SprayJsonSupport with DefaultJsonProtocol {
    implicit val personFormat = jsonFormat(Model.Person.apply, "id", "firstName", "secondaryName")
  }

  def main(args: Array[String]) {

    implicit val system = ActorSystem("my-system")
    implicit val materializer = ActorMaterializer()

    implicit val executionContext = system.dispatcher

    val start = ByteString.empty
    val sep = ByteString("\n")
    val end = ByteString.empty

    import JsonProtocol._
    implicit val jsonStreamingSupport: JsonEntityStreamingSupport = EntityStreamingSupport.json()
      .withFramingRenderer(Flow[ByteString].intersperse(start, sep, end))
      .withParallelMarshalling(parallelism = 8, unordered = false)

    val decider: Supervision.Decider = {
      case ex: Throwable => {
        println("Exception occurs")
        ex.printStackTrace()
        Supervision.Resume
      }
    }

    val persons: Source[Model.Person, NotUsed] = Source.fromIterator(
      () => (0 to 1000000).map(id => Model.Person(id = id)).iterator
    )
      .withAttributes(ActorAttributes.supervisionStrategy(decider))
      .map(p => { println(p); p })


    val route =
      path("persons") {
        get {
          complete(persons)
        }
      }

    val bindingFuture = Http().bindAndHandle(route, "localhost", 8080)

    println(s"Server online at http://localhost:8080/\nPress RETURN to stop...")
    StdIn.readLine()

    bindingFuture
      .flatMap(_.unbind())
      .onComplete(_ => {
        println("Stopping http server ...")
        system.terminate()
      })
  }
}

和客户：

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.{HttpRequest, Uri}
import akka.stream.{ActorAttributes, ActorMaterializer, Supervision}

import scala.util.{Failure, Success}

object WebClient {
  def main(args: Array[String]): Unit = {
    implicit val system = ActorSystem()
    implicit val materializer = ActorMaterializer()
    implicit val executionContext = system.dispatcher


    val request = HttpRequest(uri = Uri("http://localhost:8080/persons"))

    val response = Http().singleRequest(request)

    val attributes = ActorAttributes.withSupervisionStrategy {
      case ex: Throwable => {
        println("Exception occurs")
        ex.printStackTrace
        Supervision.Resume
      }
    }
    response.map(r => {
      r.entity.dataBytes.withAttributes(attributes)
    }).onComplete {
      case Success(db) => db.map(bs => bs.utf8String).runForeach(println)
      case Failure(ex) => ex.printStackTrace()
    }
  }
}

它适用于100,1000,1万人，但不适用于> 100 000'。看起来流有一些限制，但我找不到它

Last record has been printed by server on my local machine is (with number 79101): Person(79101,ⰷ瑐劲죗醂竜泲늎制䠸,䮳硝沢并⎗ᝨᫌꊭᐽ酡)

客户的最后记录是（编号79048）：

{"id":79048,"firstName":"췁頔䚐龫暀࡙頨捜昗㢵","secondaryName":"⏉ݾ袈庩컆◁ꄹ葪䑥Ϻ"}

也许有人知道它为什么会发生？

Answer 1

我找到了解决方案。我必须在客户端上显式添加r.entity.withoutSizeLimit（），之后全部按预期工作