Akka Соответствие сбоям и восстановлению

Question

Материал, который мне нужен, выделен жирным шрифтом.

У меня есть актер, который летает с несколькими аэрозолями HttpRequests, запросы разбиты на страницы, и актер уверен, что он последовательно записывает результаты в базу данных (последовательность важна для возобновления сканеров). Я объясняю это, потому что сейчас я не хочу изучать другие модели параллелизма. Актеру необходимо восстановить время ожидания без перезапуска.

в моем актера я следующее:

  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 
    } 
    } 
} 

Как только я нахожу правильный способ сопоставления, re в любом случае я могу получить в руки CrawlerRequest, что тайм-аут произошел с? он содержит некоторое состояние, которое мне нужно выяснить, как восстановить.

Answer 1

Необходимо указать полный путь класса case Failure (или импортировать его, я думаю).

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

Это просто оставляет доступ к запросу, связанному с таймаутом. Кажется, что карта и труба с пользовательским обработчиком сбоев необходимы при отправке запроса точки. Взгляни сейчас.

Следующие батуты тайм-аут в актер.

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 @ _, необходимо ввести класс case, чтобы поглотить избыточное сообщение об ошибке.

ОК, поэтому избавление от пипето избавляется от исключения, входящего в клиентский актер. Это также намного легче читать: 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) 
    } 
    } 
} 

Answer 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 в актера АККА в Failure (akka.actor.Status.Failure). Это отлично работает, когда вы используете шаблон ask, потому что временная подсказка актера обрабатывает akka.actor.Status.Failure для вас, но не работает с tell.

Надежда этот короткий ответ помогает :)

Удачи!

Answer 3

Если я правильно понял, вам в настоящее время не удается сопоставить AskTimeoutException.

Если это так, вы должны соответствовать case Failure(AskTimeoutException) => ... вместо case f : AskTimeoutException => ....