如何在通过Actor互操作从S3流式传输文件时处理背压

问题根因 #

你当前仅读取30行数据的核心问题是流的消费终点配置错误：
你使用了Sink.head作为流的消费Sink，该算子仅会消费上游产出的第一个元素就终止整个流。你第二个mapAsync(1)每处理完成1条Buzz数据就返回1个Done对象，当第一个Done产出后，Sink.head就会触发流终止流程。而前面mapAsync(30)的并行度配置决定了该算子会提前拉取30个元素做并行处理，所以最终仅能看到30行数据被读取。

背压逻辑验证 #

你当前通过mapAsync控制并行度的实现本身可以满足背压需求：
Akka Streams内置背压机制，下游消费速度会自动向上游传递，mapAsync(30)最多同时维持30个未完成的请求，上游不会推送更多数据，自然限制了发送给httpClientActor的请求量，不需要额外做背压处理。

修复方案 #

仅需要将终点的Sink.head替换为可消费所有上游元素的Sink即可，参考修改后的代码：

S3.download(bckt, bcktKey).map{
      case Some((file, _)) =>
        file
          .via(CsvParsing.lineScanner())
          .map(_.map(_.utf8String)).drop(1)// 丢弃表头
          .map(p => Foo(p.head, p(1)))
          .mapAsync(30) { p =>
            implicit val askTimeout: Timeout = Timeout(10 seconds)
            (httpClientActor ? p).mapTo[Buzz]
          }
          .mapAsync(1){
          case b@Buzz(_, _) =>
            (persistActor ? b).mapTo[Done]
        }.runWith(Sink.ignore) // 替换Sink.head为Sink.ignore，消费所有Done元素
}

如果需要统计处理总条数，也可以使用Sink.fold实现计数：

.runWith(Sink.fold(0)((processedCount, _) => processedCount + 1))

可选优化点 #

• 给ask操作添加异常捕获逻辑，避免单条数据请求失败直接终止整个流
• 若Actor处理能力存在波动，可搭配throttle算子实现更精细的QPS限制
• 可以配置mapAsync的监督策略，指定出错后跳过当前异常元素继续处理后续数据

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