Akka用于REST轮询

Question

我正在尝试将大型Scala + Akka + PlayMini应用程序与外部REST API连接.这个想法是定期轮询(基本上每1到10分钟)一个根URL,然后爬过子级URL以提取数据,然后将数据发送到消息队列.

我想出了两种方法:

第一路

创建一个actors层次结构以匹配API的资源路径结构.在谷歌纵横案例中,这意味着,例如

• 演员'纬度/ v1/currentLocation'民意调查https://www.googleapis.com/latitude/v1/currentLocation
• 演员'纬度/ v1 /位置'民意调查https://www.googleapis.com/latitude/v1/location
• 演员'纬度/ v1 /位置/ 1'民意调查https://www.googleapis.com/latitude/v1/location/1
• 演员'纬度/ v1 /位置/ 2'民意调查https://www.googleapis.com/latitude/v1/location/2
• 演员'纬度/ v1 /位置/ 3'民意调查https://www.googleapis.com/latitude/v1/location/3
• 等等

在这种情况下,每个actor负责定期轮询其相关资源,以及为下一级路径资源创建/删除子actor(即actor'luitude/v1/location'创建actor1,2,3等等.它通过投票https://www.googleapis.com/latitude/v1/location了解到的所有地点.

第二种方式

创建一个相同的轮询参与者池,接收由路由器负载平衡的轮询请求(包含资源路径),轮询URL一次,进行一些处理,并安排轮询请求(包括下一级资源和轮询URL) .在谷歌纵横中,这意味着例如:

1路由器,n个轮询演员.初始轮询请求https://www.googleapis.com/latitude/v1/location导致了一些新的(即时)查询请求https://www.googleapis.com/latitude/v1/location/1,HTTPS:/ /www.googleapis.com/latitude/v1/location/2等,以及相同资源的一个(延迟)轮询请求,即https://www.googleapis.com/latitude/v1/location.

我已经实现了两种解决方案,并且不能立即观察到任何相关的性能差异,至少不是我感兴趣的API和轮询频率.我发现第一种方法更容易推理,也许更容易使用系统.scheduler.schedule(...)比第二种方法(我需要scheduleOnce(...)).此外,假设资源嵌套在几个级别并且有些短暂(例如,可以在每次轮询之间添加/删除多个资源),akka的生命周期管理可以在第一种情况下轻松杀死整个分支.第二种方法应该(理论上)更快,代码更容易编写.

我的问题是:

1. 什么方法似乎是最好的(在性能,可扩展性,代码复杂性等方面)？
2. 你认为这两种方法的设计都有问题(尤其是第一种方法)吗？
3. 有没有人试图实现类似的任何东西？怎么做的？

谢谢!

Answer 1

为什么不创建一个主轮询器，然后按计划启动异步资源请求？

我不是使用 Akka 的专家，但我尝试了一下：

循环访问要获取的资源列表的 poller 对象：

import akka.util.duration._
import akka.actor._
import play.api.Play.current
import play.api.libs.concurrent.Akka

object Poller {
  val poller = Akka.system.actorOf(Props(new Actor {
    def receive = {
      case x: String => Akka.system.actorOf(Props[ActingSpider], name=x.filter(_.isLetterOrDigit)) ! x
    }
  }))

  def start(l: List[String]): List[Cancellable] =
    l.map(Akka.system.scheduler.schedule(3 seconds, 3 seconds, poller, _))

  def stop(c: Cancellable) {c.cancel()}
}

异步读取资源并触发更多异步读取的参与者。如果更友善的话，您可以将消息发送安排在时间表上，而不是立即致电：

import akka.actor.{Props, Actor}
import java.io.File

class ActingSpider extends Actor {
  import context._
  def receive = {
    case name: String => {
      println("reading " + name)
      new File(name) match {
        case f if f.exists() => spider(f)
        case _ => println("File not found")
      }
      context.stop(self)
    }
  }

  def spider(file: File) {
    io.Source.fromFile(file).getLines().foreach(l => {
      val k = actorOf(Props[ActingSpider], name=l.filter(_.isLetterOrDigit))
      k ! l
    })
  }
}