akka 流 onComplete{} 没有结束

Question

我正在尝试使用 akka 流读取多个文件。处理结果并将处理结果导出到文件中。 程序卡在 onComplete{} 中。在我手动停止程序之前不会导出结果。

我尝试了以下代码。在我停止程序之前不会执行导出。即使在停止程序之前，导出之前或之后的任何其他处理都会以任何方式执行。

// this function reads multiples files 
@throws[FileNotFoundException]
def concatFilesAkka(path : String, date : String, numberOfDays : 
Int) : Future[Seq[String]] = {

implicit val system = ActorSystem("Sys")
val settings = ActorMaterializerSettings(system)
implicit val materializer = ActorMaterializer(settings)

val formatter = DateTimeFormatter.ofPattern("yyyyMMdd");
val formattedDate = LocalDate.parse(date, formatter);

def files = {.. }

val result = Source(files).flatMapConcat(filename =>

  FileIO.fromPath(Paths.get(filename))
    .via(Framing.delimiter(ByteString("\n"), 256, allowTruncation = 
true).map(_.utf8String))
).toMat(Sink.seq)(Keep.right)
  .run()

result
}





// processing 
def process(date: String, inPath: String,  outPath: String, 
..): Unit = {

implicit val system = ActorSystem("Sys")
val settings = ActorMaterializerSettings(system)
implicit val materializer = ActorMaterializer(settings)


  val a =  concatFilesAkka(inPath,date, numberOfDays)
  a.onComplete(x => {
  var transactions = x.get.map(line => line.split('|')).toList

 val groupedByMagasin = transactions.filter(x => x(3) != 
 "0").groupBy(x 
  => x(2)).foreach(x => {

  .......
  .......
val top100VenteGlobale: List[(String, Int)] = Nil
val top100CaGlobale: List[(String, Int)] = Nil

...
...



})

//exporting top 100 vente global et top 100 ca global
export(top100VenteGlobale, outPath + "top_100_vente_GLOBAL_" + 
date })  
}

}

Answer 1

如果您将其作为脚本运行，则必须手动停止/终止 Actor 系统或存在系统。