从 Akka Actor 那里获取消息

Question

我构建了一个定期查询 API 的 Akka actor，如下所示：

  val cancellable =
    system.scheduler.schedule(0 milliseconds,
      5 seconds,
      actor,
      QueryController(1))

Actor，本质上是：

object UpdateStatistics {
  /**
   * Query the controller for the given switch Id
   *
   * @param dpId Switch's Id
   */
  case class QueryController(dpId: Int)
  case object Stop

  def props: Props = Props[UpdateStatistics]
}

class UpdateStatistics extends Actor with akka.actor.ActorLogging {
  import UpdateStatistics._

  def receive = {

    case QueryController(id) =>
      import context.dispatcher
      log.info(s"Receiving request to query controller")
      Future { FlowCollector.getSwitchFlows(1) } onComplete {
        f => self ! f.get
      }
    case Stop =>
      log.info(s"Shuting down")
      context stop self
    case json: JValue =>
      log.info("Getting json response, computing features...")
      val features = FeatureExtractor.getFeatures(json)
      log.debug(s"Features: $features")
      sender ! features
    case x =>
      log.warning("Received unknown message: {}", x)
  }
}

我想做的是从UpdateStatistics 演员那里得到json:Jvalue 消息。阅读Akka docs 我认为这可能有用：

  implicit val i = inbox()
  i.select() {
     case x => println(s"Valor Devuelto $x")
  }
  println(i receive(2.second))

但我不知道如何修改UpdateStatisticsactor 以便将结果发送到上面的收件箱。

我在文档中阅读的另一个选项是 event streams。

但我认为这不是正确的方法。

有没有办法实现我想做的事情？还是我需要使用第二个Actor 来向其发送JSON 响应？

Answer 1

您可能正在 AKKA 中寻找 ask 模式。这将允许您向发件人返回一个值。

import akka.pattern.ask
import akka.util.duration._

implicit val timeout = Timeout(5 seconds)

val future = actor ? QueryController(1)    
val result = Await.result(future, timeout.duration).asInstanceOf[JValue]

println(result)

要完成这项工作，您需要将响应发送到原始sender，而不是self。此外，您应该注意将来在处理消息时关闭sender 的危险。