Akka Stream添加Throttle后仅输出首行即终止问题排查

问题现象 #

使用Akka Stream流式读取文件内容并打印到控制台：

• 未添加限流时，所有文本行可快速正常输出；
• 添加throttle(1, 1.second, 1, ThrottleMode.shaping)（每秒输出1行）后，仅首行被打印，流随即终止。

核心原因 #

Akka Stream的处理逻辑是异步执行的，而main方法执行完毕后会直接触发JVM退出。添加限流后，流的后续元素需要延迟处理，但此时ActorSystem已随JVM终止而关闭，导致剩余元素无法被处理。
未加限流时，文件读取与打印速度极快，所有元素在main方法退出前就完成处理，因此能正常输出全部内容。

修正方案 #

需要等待流处理完成的Future，确保所有元素都被处理后再终止JVM。以下是两种可行方式：

方式1：阻塞等待流完成（简单直接） #

通过Await.result阻塞main线程，直到流处理完成，再手动终止ActorSystem：

import scala.concurrent.Await
import scala.concurrent.duration._

def main(args: Array[String]): Unit = {

  val eSource: Source[ByteString, Future[IOResult]] = 
    rawDataStream(conf.eventsFilePath)

  val eFlow = 
    Flow[ByteString]
      .via(Framing.delimiter(ByteString(System.lineSeparator), 10000))
      .map(bs => bs.utf8String)
      .throttle(1, 1.second, 1, ThrottleMode.shaping)

  val eSink = Sink.foreach(println)

  // 获取流处理结果的Future
  val streamResult: Future[IOResult] = eSource.via(eFlow).runWith(eSink)

  // 阻塞等待流完成，超时时间需根据文件行数和限流速度调整（示例设为10秒）
  Await.result(streamResult, 10.seconds)

  // 流处理完成后终止ActorSystem
  AkkaStreamUtils.defaultActorSystem.terminate()
}

方式2：非阻塞监听流完成 #

通过onComplete监听流处理的Future，完成后自动终止ActorSystem，无需阻塞main线程：

import scala.concurrent.ExecutionContext.Implicits.global

def main(args: Array[String]): Unit = {

  val eSource: Source[ByteString, Future[IOResult]] = 
    rawDataStream(conf.eventsFilePath)

  val eFlow = 
    Flow[ByteString]
      .via(Framing.delimiter(ByteString(System.lineSeparator), 10000))
      .map(bs => bs.utf8String)
      .throttle(1, 1.second, 1, ThrottleMode.shaping)

  val eSink = Sink.foreach(println)

  val streamResult: Future[IOResult] = eSource.via(eFlow).runWith(eSink)

  // 流处理完成后终止ActorSystem
  streamResult.onComplete { _ =>
    AkkaStreamUtils.defaultActorSystem.terminate()
  }
}

注意事项 #

• 方式1中的超时时间需合理设置：比如文件有5行、限流每秒1行，超时时间至少要大于5秒，避免提前中断流处理。
• 确保ActorSystem是显式终止的，避免资源泄漏。

内容的提问来源于stack exchange，提问作者nicholsonjf