scala – 在Akka / Spray上的油门HTTP请求

我在 Scala中使用Akka演员从外部服务(HTTP获取请求)下载资源.外部服务的响应是 JSON,我必须使用分页(提供程序很慢).我想在10个线程中同时下载所有分页结果.我使用这样的URL下载块： http://service.com/itmes?limit=50&offset=1000

我创建了以下管道：

ScatterActor => RoundRobinPool[10](LoadChunkActor) => Aggreator

ScatterActor需要总共下载的项目数量并将其划分成块.我创建了10个LoadChunkActor来同时处理任务.

override def receive: Receive = {
    case LoadMessage(limit) =>
    val offsets: IndexedSeq[Int] = 0 until limit by chunkSize
    offsets.foreach(offset => context.system.actorSelection(pipe) !
    LoadMessage(chunkSize,offset))
 }

LoadChunkActor使用Spray发送请求.演员看起来像这样：

val pipeline = sendReceive ~> unmarshal[List[Items]]
override def receive: Receive = {
  case LoadMessage(limit,offset) =>
    val uri: String = s"http://service.com/items?limit=50&offset=$offset"
    val responseFuture = pipeline {Get(uri)}
    responseFuture onComplete {
      case Success(items) => aggregator ! Loaded(items)
    }
 }

正如你可以看到的,LoadChunkActor正在从外部服务请求块,并添加回调来运行onComplete.演员现在准备再发一个消息,他正在请求另一个大块. Spray正在使用非阻塞API来下载块.结果外部服务充斥着我的请求,我得到超时.

如何安排任务列表,但是我想同时处理最多10个？

ScatterActor (10000x messages) => 
  ThrottleActor => LoadChunkActor => ThrottleMonitorActor => Aggregator
         ^                                    |
         |<--------WorkDoneMessage------------|