Akka 流 Source.repeat 在 100 个请求后停止答案

【问题标题】：Akka streams Source.repeat stops after 100 requestsAkka 流 Source.repeat 在 100 个请求后停止
【发布时间】：2019-01-08 01:23:09
【问题描述】：

我正在开发以下流处理系统，以从一个来源抓取帧，处理并发送到另一个来源。我通过他们的 scapa api 使用了 akka-streams 和 akka-http 的组合。管道非常短，但我似乎无法找到系统在向端点发出 100 个请求后决定停止的位置。

object frameProcessor extends App {
  implicit val system: ActorSystem = ActorSystem("VideoStreamProcessor")
  val decider: Supervision.Decider = _ => Supervision.Restart
  implicit val materializer: ActorMaterializer = ActorMaterializer()
  implicit val dispatcher: ExecutionContextExecutor = system.dispatcher
  val http = Http(system)
  val sourceConnectionFlow: Flow[HttpRequest, HttpResponse, Future[Http.OutgoingConnection]] = http.outgoingConnection(sourceUri)

  val byteFlow: Flow[HttpResponse, Future[ByteString], NotUsed] =
    Flow[HttpResponse].map(_.entity.dataBytes.runFold(ByteString.empty)(_ ++ _))

  Source.repeat(HttpRequest(uri = sourceUri))
    .via(sourceConnectionFlow)
    .via(byteFlow)
    .map(postFrame)
    .runWith(Sink.ignore)
    .onComplete(_ => system.terminate())

  def postFrame(imageBytes: Future[ByteString]): Unit = {
    imageBytes.onComplete{
      case Success(res) => system.log.info(s"post frame. ${res.length} bytes")
      case Failure(_) => system.log.error("failed to post image!")
    }
  }
}

作为参考，我使用的是akka-streams 版本2.5.19 和akka-http 版本10.1.7。没有抛出错误，帧来自的源服务器上没有错误代码，程序以错误代码0退出。

我的application.conf如下：

logging = "DEBUG"

始终处理 100 个单位。

谢谢！

编辑

像这样将日志记录添加到流中

.onComplete{
  case Success(res) => {
    system.log.info(res.toString)
    system.terminate()
  }
  case Failure(res) => {
    system.log.error(res.getMessage)
    system.terminate()
  }

}

收到连接重置异常，但这不一致。流以Done 结束。

编辑 2

使用 .mapAsync(1)(postFrame) 在恰好 100 个请求后，我得到相同的 Success(Done)。此外，当我检查 nginx 服务器 access.log 和 error.log 时，只有 200 响应。

我必须修改postFrame 如下运行mapAsync

def postFrame(imageBytes: Future[ByteString]): Future[Unit] = {
  imageBytes.onComplete{
    case Success(res) => system.log.info(s"post frame. ${res.length} bytes")
    case Failure(_) => system.log.error("failed to post image!")
  }
  Future(Unit)
}

【问题讨论】：

如果将日志记录添加到.onComplete(_ => system.terminate()) 会发生什么？可能是服务器在 100 帧后才停止发送数据。
感谢您的提示。看起来连接已被对等主机重置。我认为使用重新启动决定器会导致流重新启动，但事实并非如此。我会在哪里发现连接重置错误？不是在这个例子中，但在我的代码中，我使用了RestartSource.onFailureWithBackoff，但它仍然会出错。感谢您的回复
运行几次，并不总是出现连接重置错误。我刚收到Success(Done)
@Andrew 如果你使用mapAsync 而不是map 会发生什么？
我刚刚注意到的一件事（它可能只是在您的问题代码中，而不是在您的真实代码中），但是您的 decider 是否被使用过？我认为它应该被传递给ActorMaterializer构造函数。

标签： scala akka akka-stream akka-http

【解决方案1】：

我相信我在Akka docs using delayed restarts with a backoff operator 上找到了答案。我没有直接从不稳定的远程连接采购，而是使用RestartSource.withBackoff 和not RestartSource.onFailureWithBackoff。修改后的流看起来像；

val restartSource = RestartSource.withBackoff(
  minBackoff = 100.milliseconds,
  maxBackoff = 1.seconds,
  randomFactor = 0.2
){ () =>
  Source.single(HttpRequest(uri = sourceUri))
    .via(sourceConnectionFlow)
    .via(byteFlow)
    .mapAsync(1)(postFrame)
}
restartSource
  .runWith(Sink.ignore)
  .onComplete{
    x => {
      println(x)
      system.terminate()
    }
 }

我无法找到问题的根源，但似乎这会奏效。

【讨论】：