如何突然停止 akka 流 Runnable Graph？答案

【问题标题】：How to abruptly stop an akka stream Runnable Graph?如何突然停止 akka 流 Runnable Graph？
【发布时间】：2016-08-03 14:53:05
【问题描述】：

我无法弄清楚如何立即停止 akka 流 Runnable Graph？如何使用 killswitch 来实现这一点？我开始使用 akka 流才几天。就我而言，我正在从文件中读取行并在流中执行一些操作并写入接收器。我想要做的是，随时停止读取文件，我希望这可能会停止整个运行图。对此的任何想法将不胜感激。

提前致谢。

【问题讨论】：

标签： scala io akka akka-stream scala-streams

【解决方案1】：

从 Akka Streams 2.4.3 开始，有一种优雅的方式可以通过 KillSwitch 从外部停止流。

考虑以下示例，它会在 10 秒后停止流式传输。

object ExampleStopStream extends App {

  implicit val system = ActorSystem("streams")
  implicit val materializer = ActorMaterializer()

  import system.dispatcher

  val source = Source.
    fromIterator(() => Iterator.continually(Random.nextInt(100))).
    delay(500.millis, DelayOverflowStrategy.dropHead)
  val square = Flow[Int].map(x => x * x)
  val sink = Sink.foreach(println)

  val (killSwitch, done) =
    source.via(square).
    viaMat(KillSwitches.single)(Keep.right).
    toMat(sink)(Keep.both).run()

  system.scheduler.scheduleOnce(10.seconds) {
    println("Shutting down...")
    killSwitch.shutdown()
  }

  done.foreach { _ =>
    println("I'm done")
    Await.result(system.terminate(), 1.seconds)
  }

}

【讨论】：

见下例代码：如何停止下图，取消读取大文件？ RunnableGraph.fromGraph(GraphDSL.create() {implicit builder => import GraphDSL.Implicits._ // Source from file val A: Outlet[String] = builder.add(readFileLineByLine).out //将每一行写入文件 val E : Inlet[String] = builder.add(writeLinesToFile).in A ~> E ClosedShape }).run()
您正在提供示例，以便您在开始之前知道何时停止系统，但就我而言，我需要随时停止正在运行的系统。
我看到，从 akka 流测试我开始工作了： val switch1 = KillSwitches.shared("switch") val upstream = RunnableGraph.fromGraph(GraphDSL.create() {implicit builder => import GraphDSL.Implicits._ // Source from file val A: Outlet[String] = builder.add(readFileLineByLine).out //将每一行写入文件 val E: Inlet[String] = builder.add(writeLinesToFile).in A .via(switch1.flow) ~> E ClosedShape }).run() switch1.shutdown()

【解决方案2】：

一种方式有一个服务或shutdownhookup，可以调用graphcancellable

val graph=
    Source.tick(FiniteDuration(0,TimeUnit.SECONDS), FiniteDuration(1,TimeUnit.SECONDS), Random.nextInt).to(Sink.foreach(println))
  val cancellable=graph.run()

  cancellable.cancel

cancellable.cancel 可以是 ActorSystem.registerOnTermination 的一部分

【讨论】：