1

我正在为我的雇主构建一个小工具，并尝试使用 Akka 2 + Scala 来实现它。

我正在与之交互的系统之一有一个 SOAP api，它有一个有趣的实现：

- - 要求 - - - -

客户端----请求---->服务器

客户端<---请求确认-----服务器

- - 回复 - - -

客户端<---响应----服务器

Client ----Resp Ack---->Server

跟踪请求是通过 req-ack 发送的任务 ID 完成的。

我的问题是 SO 社区，任何有 akka 经验的人都如下：

  • 对于这种特定场景，使用 akka 处理 http 消息的理想方法是什么？我已经看到了 Akka-Mist 的示例，看起来它已经停止支持 play-mini，然后是 akka-camel，我似乎无法找到作为 akka 2.0.3 一部分的库分布，所以我有点困惑我应该采取什么方法。是否有推荐的方法将 servlet 的行为封装在 akka actor 中？

谢谢大家。

4

1 回答 1

3

您的问题非常开放，因此我将对您的要求做出一些假设。

请求假设：某个事件会产生很多请求。

创建一个包含包装 async-http-client (https://github.com/sonatype/async-http-client) 的参与者的路由器。一旦 async-http-client 完成一个请求，它就会向协调参与者发送一条消息，该消息的 ID 包含在 Req Ack 中。协调参与者将聚合 ID。

以下大部分是伪代码，但它与真正的 API 足够接近，您应该能够找出缺失的部分。

case class Request(url:String)
case class IDReceived(id:String)

case class RequestingActor extends Actor {

    override def receive = {
        case Request(url) => {
            //set up the async client and run the request
            //since it's async the client will not block and this actor can continue making generating more requests
        }
    }
}


class AsyncHttpClientResponseHandler(aggregationActor:ActorRef) extends SomeAsyncClientHandler {

    //Override the necessary Handler methods.

    override def onComplete = {
        aggregationActor ! IDReceived(//get the id from the response)
    }
}

class SomeEventHandlerClass { 

    val requestRouter = actorSystem.actorOf(Props[RequestingActor].withRouter(FromConfig(//maybe you've configured a round-robin router)), requestRouterName)

    def onEvent(url:String) {
        requestRouter ! Request(url)
    }

}

case class AggregationActor extends Actor {

    val idStorage = //some form of storage, maybe a Map if other information about the task ID needs to be stored as well.  Map(id -> other information)

    override def receive = {
        case IDReceived(id) => //add the ID to idStorage
    }
}

响应假设：您需要对响应中包含的数据做一些事情，然后将 ID 标记为完整。HTTP 前端只需要处理这一组消息。

不要试图找到一个与 Akka 集成的框架，只需使用一个简单的 HTTP 前端，它将消息发送到您创建的 Akka 网络。我无法想象使用 play-mini 获得的任何优势，而且我认为它会掩盖一些描绘工作分离和控制的界限。情况并非总是如此，但鉴于您的问题缺乏要求，我没有其他可依据的意见。

case class ResponseHandlingActor extends Actor {

    val aggregationActor = actorSystem.actorFor(//the name of the aggregation router within the Actor System)

    override def receive = {
        case Response(data) => {
            //do something with the data.  If the data manipulation is blocking or long running it may warrant its own network of actors.
            aggregationActor ! ResponseReceived(//get the id from the response)
        }
    }
}

class ScalatraFrontEnd() extends ScalatraServlet {

    val responseRouter = actorSystem.actorOf(Props[RequestingActor].withRouter(FromConfig(//maybe you've configured a round-robin router)), requestRouterName)

    post("/response") {
        responseRouter ! Response(//data from the response)
    }

}

创建一个与此类似的系统可以很好地分离关注点，可以很容易地推断处理发生在哪里，并有足够的空间来扩展或扩展系统。

于 2012-09-26T12:59:43.437 回答