关闭Akka流以进行资源清理

Question

当使用Akka流时，是否有任何方法关闭/关闭资源清理不再需要的流？

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

示例用法：

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

有办法关闭流吗？

Answer 1

您可以在独立的Stream上运行ActorMaterializer，并在一定时间后调用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)
}

类似地，您可以返回ActorMaterializer而不是物化流，并根据时间以外的某些外部条件关闭ActorMaterializer。