關閉資源清理的Akka流

Question

當使用Akka Streams時，有什麼方法可以關閉/關閉不再需要資源清理的流嗎？

編輯：當源由無限流組成時，它可能永遠不會完成，我想在完成源之前停止它。

用法示例：

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

有沒有一種方法來關閉流？

Answer 1

您可以運行在一個獨立的ActorMaterializer的Stream和一段時間後調用shutdown的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一些外部條件，而不是時間。