使用Akka和Scalatra

Question

我的目标是为我的小部件构建高度并发的后端。我目前正在将后端展示为一个Web服务，该服务接收运行特定窗口小部件（使用Scalatra）的请求，从数据库中获取窗口小部件的代码并将其运行在一个actor中（使用Akka），然后使用结果进行回复。所以，想象我在做这样的事情：

get("/run:id") { 
    ... 
    val actor = Actor.actorOf("...").start 
    val result = actor !! (("Run",id), 10000) 
    ... 
} 

现在，我相信这是不是最好的并发解决方案，我应该以某种方式结合起来监听请求，并以一个演员的实施运行的微件。你如何设计这个最大并发性？谢谢。

Answer 1

您可以在akka启动文件或您自己的ServletContextListener中启动您的演员，以便它们在不受绑定到servlet的情况下启动。 然后，您可以使用akka注册表查找它们。

Actor.registry.actorFor[MyActor] foreach { _ !! (("Run",id), 10000) } 

除此之外，此刻还没有真正的akka​​与scalatra的整合。 因此，直到现在，您所能做的最好的方式是对一群演员使用阻止请求。

我不确定，但我不会为每个请求产生一个actor，而是有一个可以发送这些请求的小部件actor的池。如果您使用管理员层次结构，那么您可以使用管理员来调整池的大小，如果它太大或太小。

class MyContextListener extends ServletContextListener { 

    def contextInitialized(sce: ServletContextEvent) { 
    val factory = SupervisorFactory(
     SupervisorConfig(
     OneForOneStrategy(List(classOf[Exception]), 3, 1000), 
     Supervise(actorOf[WidgetPoolSupervisor], Permanent) 
    } 

    def contextDestroyed(sce: ServletContextEvent) { 
    Actor.registry.shutdownAll() 
    } 
} 

class WidgetPoolSupervisor extends Actor { 

    self.faultHandler = OneForOneStrategy(List(classOf[Exception]), 3, 1000) 

    override def preStart() { 
    (1 to 5) foreach { _ => 
     self.spawnLink[MyWidgetProcessor] 
    } 
    Scheduler.schedule(self, 'checkPoolSize, 5, 5, TimeUnit.MINUTES) 
    } 

    protected def receive = { 
    case 'checkPoolSize => { 
     //implement logic that checks how quick the actors respond and if 
     //it takes to long add some actors to the pool. 
     //as a bonus you can keep downsizing the actor pool until it reaches 1 
     //or until the message starts returning too late. 
    } 
    } 
} 

class ScalatraApp extends ScalatraServlet { 

    get("/run/:id") { 
    // the !! construct should not appear anywhere else in your code except 
    // in the scalatra action. You don't want to block anywhere else, but in a 
    // scalatra action it's ok as the web request itself is synchronous too and needs to 
    // to wait for the full response to have come back anyway. 
    Actor.registry.actorFor[MyWidgetProcessor] foreach { 
     _ !! ((Run, id), 10000) 
    } getOrElse { 
     throw new HeyIExpectedAResultException() 
    } 
    } 
} 

请不要考虑上面的代码，如发生这种情况看起来像斯卡拉伪代码，我只是想说明这个概念。