Akka 匹配失败和恢复答案

【问题标题】：Akka matching Failures, and recoveryAkka 匹配失败和恢复
【发布时间】：2013-12-08 11:04:10
【问题描述】：

我需要帮助的内容以粗体显示。

我有一个执行多个喷射 HttpRequest 的参与者，请求是分页的，并且参与者确保它按顺序将结果写入数据库（顺序对于恢复爬虫很重要）。我解释这一点是因为我目前不想探索其他并发模式。演员需要从超时中恢复而不重新启动。

在我的演员中，我有以下内容：

            case f : Failure => {
                system.log.error("faiure")
                system.log.error(s"$f")
                system.shutdown()
            }
            case f : AskTimeoutException => {
                system.log.error("faiure")
                system.log.error(s"$f")
                system.shutdown()
            }
            case msg @ _ => {

                system.log.error("Unexpected message in harvest")
                system.log.error(s"${msg}")
                system.shutdown()
            }

但我无法正确匹配：

[ERROR] [11/23/2013 14:58:10.694] [Crawler-akka.actor.default-dispatcher-3] [ActorSystem(Crawler)] Unexpected message in harvest
[ERROR] [11/23/2013 14:58:10.694] [Crawler-akka.actor.default-dispatcher-3] [ActorSystem(Crawler)] Failure(akka.pattern.AskTimeoutException: Timed out)

我的调度如下：

abstract class CrawlerActor extends Actor {
  private implicit val timeout: Timeout = 20.seconds
  import context._
  def dispatchRequest(node: CNode) {
    val reqFut = (System.requester ? CrawlerRequest(node,Get(node.url))).map(r=> CrawlerResponse(node,r.asInstanceOf[HttpResponse]))
    reqFut pipeTo self
  }


class CrawlerRequester extends Actor  {
  import context._
  val throttler = context.actorOf(Props(classOf[TimerBasedThrottler],System.Config.request_rate),"throttler")
  throttler ! SetTarget(Some(IO(Http).actorRef))

  def receive : Receive = {
    case CrawlerRequest(type_,request) => {
      throttler forward request
    }
  }
}

一旦我找到了正确的匹配方式，我是否可以掌握发生超时的 CrawlerRequest ？它包含一些我需要弄清楚如何恢复的状态。

【问题讨论】：

标签： scala akka actor spray

【解决方案1】：

需要输入Failure case类的完整路径，（或者我猜是导入）。

case f: akka.actor.Status.Failure => {
                system.log.error("faiure")
                system.log.error(s"${f.cause}")
                system.shutdown()
            }

剩下的就是处理与超时相关的请求。似乎在点请求分派时需要带有自定义故障处理程序的地图和管道。现在调查一下。

下面的蹦床将超时放入actor中。

case class CrawlerRequestTimeout(request: CrawlerRequest)
abstract class CrawlerActor extends Actor {
  private implicit val timeout: Timeout = 20.seconds
  import context._
  def dispatchRequest(node: CNode) {
    val req =  CrawlerRequest(node,Get(node.url))
    val reqFut = (System.requester ? req).map(r=> CrawlerResponse(node,r.asInstanceOf[HttpResponse]))

    reqFut onFailure {
        case te: akka.pattern.AskTimeoutException => self ! CrawlerRequestTimeout(req)
    }
    reqFut pipeTo self
  }
}

匹配：

 case timeout : CrawlerRequestTimeout => {
                println("boom")
                system.shutdown()
            }

需要找到一种方法来抑制异常，但它仍在触发。也许压制不是真正的问题，验证。

不，抑制是一个问题，否则异常会渗透到 msg @ _，需要放入一个案例类来吸收多余的失败消息。

好的，因此摆脱 pipeto 可以摆脱进入客户端 actor 的异常。它也更容易阅读：D

abstract class CrawlerActor extends Actor {
  private implicit val timeout: Timeout = 20.seconds
  import context._
  def dispatchRequest(node: CNode) {
    val req =  CrawlerRequest(node,Get(node.url))
    val reqFut = (System.requester ? req)

    reqFut onFailure {
        case te: akka.pattern.AskTimeoutException => self ! CrawlerRequestTimeout(req)
    }
    reqFut onSuccess {
        case r: HttpResponse => self ! CrawlerResponse(node,r)
    }
  }
}

【讨论】：

【解决方案2】：

如果您使用pipeTo回复tell发送的消息，就会出现这种情况。

例如：

in actorA: actorB ! message
in actorB: message => doStuff pipeTo sender
in actorA: receives not 'scala.util.Failure', but 'akka.actor.Status.Failure'

pipeTo 中的附加逻辑是将Try 的Failure 转换为akka 的actor Failure (akka.actor.Status.Failure)。这在您使用 ask 模式时效果很好，因为临时要求演员为您处理 akka.actor.Status.Failure，但不适用于 tell。

希望这个简短的回答有帮助:)

祝你好运！

【讨论】：

【解决方案3】：

如果我理解正确，您目前没有成功匹配AskTimeoutException。

如果是这样，您应该匹配 case Failure(AskTimeoutException) => ... 而不是 case f : AskTimeoutException => ...。

【讨论】：