如何突然停止 akka 流 Runnable Graph?
How to abruptly stop an akka stream Runnable Graph?
我不知道如何立即停止 akka 流 Runnable Graph?如何使用 killswitch 来实现这一点?我开始使用 akka 流才几天。在我的例子中,我正在从文件中读取行并在流中执行一些操作并写入接收器。我想要做的是,在我想要的时候立即停止读取文件,我希望这应该可以停止整个 运行 图。对此有任何想法将不胜感激。
提前致谢。
一种方法有一个服务或 shutdownhookup 可以调用图形可取消
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
的一部分
从 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)
}
}
我不知道如何立即停止 akka 流 Runnable Graph?如何使用 killswitch 来实现这一点?我开始使用 akka 流才几天。在我的例子中,我正在从文件中读取行并在流中执行一些操作并写入接收器。我想要做的是,在我想要的时候立即停止读取文件,我希望这应该可以停止整个 运行 图。对此有任何想法将不胜感激。
提前致谢。
一种方法有一个服务或 shutdownhookup 可以调用图形可取消
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
的一部分从 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)
}
}