10

我在 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 上运行的回调。Actor 现在准备好接收另一条消息并且他正在请求另一个块。Spray 正在使用非阻塞 API 来下载块。结果外部服务被我的请求淹没了，我得到了超时。

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

4

2 回答 2

3

我创建了以下解决方案（类似于拉http://www.michaelpollmeier.com/akka-work-pulling-pattern/：

ScatterActor (10000x messages) => 
  ThrottleActor => LoadChunkActor => ThrottleMonitorActor => Aggregator
         ^                                    |
         |<--------WorkDoneMessage------------|
  1. ThrottleActor 将消息发布到 ListBuffer 并向 LoadChunkActor 发送最多 N 条消息。
  2. 当 LoadChunkActor 通过 ThrottleMonitorActor 向 Aggregator 发送消息时。
  3. ThrottleMonitorActor 向 ThrottleActor 发送确认。
  4. ThrottleActor 将下一条消息发送到 LoadChunkActor。
于 2014-08-08T10:24:45.330 回答
1

从项目adhoclabs/akka-http-contrib中，你现在（2016 年 7 月，两年后）来自Yeghishe Piruzyan的scala.co.adhoclabs.akka.http.contrib.throttle包。

请参阅“ Akka Http 请求限制”

implicit val throttleSettings = MetricThrottleSettings.fromConfig

Http().bindAndHandle(
  throttle.apply(routes),
  httpInterface,
  httpPort
)
于 2016-07-25T08:52:06.183 回答