关闭 Akka Stream 以进行资源清理

Question

使用 Akka Streams 时，有什么方法可以 close/shutdown 不再需要资源清理的流吗？

编辑：当源由无限流组成时，它可能永远不会完成，我想在完成源之前停止它。

用法示例：

Source.from(publisher)
      .map((p) -> p)
      .to(Sink.ignore())
      .run(materializer)

有没有办法关闭流？

Answer 1

您可以在独立 ActorMaterializer 上运行 Stream 并在一段时间后在 ActorMaterializer 上调用关闭：

val actorSystem = ActorSystem()

val temporaryStream = {

  val localMat = ActorMaterializer()(actorSystem)

  import actorSystem.dispatcher
  actorSystem.scheduler.scheduleOnce(10 minutes) { localMat.shutdown() }

  Source.from(publisher)
        .map((p) -> p)
        .to(Sink.ignore())
        .run()(localMat)
}

同样，您可以 return ActorMaterializer 而不是物化流，并根据时间以外的一些外部条件关闭 ActorMaterializer。

关闭 Akka Stream 以进行资源清理

Closing an Akka Stream for resource cleanup

akka

apache-kafka

akka-stream