Pekko中throttle与Sink.foreachAsync结合的异步限流问题

你的代码里throttle是在Source端控制元素流入Sink的速率，而非任务完成的速率；而Sink.foreachAsync(32)的并行度设为32，意味着只要有元素到达Sink，就会立即启动对应的任务（调用task()），不管之前的任务是否完成。所以即使throttle每秒只放行1个元素，只要并行度允许，就会连续启动任务，导致你看到多条"starting task"输出。

你可以在Sink端实现严格的速率控制，以下是两种适配不同需求的方案：

方案1：串行处理+固定延迟（确保前一个任务完成后再发起下一个） #

如果你的需求是必须等前一个API请求完成后，间隔固定时间再发起下一个，可以用Sink.foldAsync实现串行异步处理，严格控制任务完成的速率：

import akka.actor.ActorSystem
import akka.pattern.after
import scala.concurrent.{Future, Promise}
import scala.concurrent.duration._

implicit val system: ActorSystem = ActorSystem("TaskThrottling")
implicit val ec = system.dispatcher

type Task = () => Future[Any]

val q = Source.queue[Task](4, OverflowStrategy.backpressure, 256)
  .to(Sink.foldAsync(())) { (_, task) =>
    println("starting task")
    task().map(v => {
      println(s"got $v")
    }).flatMap(_ => 
      // 任务完成后等待1秒，再处理下一个任务
      after(1.second, system.scheduler)(Future.successful(()))
    )
  }
  .run()

// 测试代码
val p1 = Promise[String]()
val p2 = Promise[String]()
val p3 = Promise[String]()
q.offer(() => p1.future); q.offer(() => p2.future); q.offer(() => p3.future);

// 手动完成Promise模拟任务结束
p1.success("task1 done")
p2.success("task2 done")
p3.success("task3 done")

这个方案中，Sink.foldAsync会串行处理每个任务：只有前一个任务的Future完成并等待指定延迟后，才会处理下一个任务，完全符合你控制任务完成速率的需求。

方案2：并发控制+速率限流（允许少量并发但整体速率不超限） #

如果你需要允许少量并发请求，但整体完成速率不超过设定值，可以结合mapAsync（控制并发数）和throttle（控制整体完成速率）：

import akka.actor.ActorSystem
import scala.concurrent.{Future, Promise}
import scala.concurrent.duration._

implicit val system: ActorSystem = ActorSystem("TaskThrottling")
implicit val ec = system.dispatcher

type Task = () => Future[Any]

val q = Source.queue[Task](4, OverflowStrategy.backpressure, 256)
  .mapAsync(3) { task =>  // 最多同时处理3个任务
    println("starting task")
    task().map(v => {
      println(s"got $v")
      v
    })
  }
  .throttle(1, 1.second, 1, _ => 1, ThrottleMode.shaping)  // 每秒最多完成1个任务
  .to(Sink.ignore)
  .run()

// 测试代码
val p1 = Promise[String]()
val p2 = Promise[String]()
val p3 = Promise[String]()
q.offer(() => p1.future); q.offer(() => p2.future); q.offer(() => p3.future);

p1.success("task1 done")
p2.success("task2 done")
p3.success("task3 done")

这里throttle放在mapAsync之后，作用于任务完成后的结果，既能控制整体完成速率不超过每秒1个，又允许最多3个任务并发执行。

• 若要在Sink端实现限流，Sink.foldAsync是最直接的方式，它天然支持串行异步处理，能严格控制任务的发起和完成顺序。
• 避免用foreachAsync做限流，它的设计目标是处理并发任务，不保证任务完成的顺序和速率限制。

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