如何处理Akka执行元内部的WS调用超时

Question

我有以下actor向WebService发送请求：

class VigiaActor extends akka.actor.Actor {
  val log = Logging(context.system, this)

  context.setReceiveTimeout(5 seconds)

  import VigiaActor._
  def receive = {
    case ObraExists(numero: String, unidadeGestora: String) =>
      WS.url(baseURL + s"""/obras/exists/$unidadeGestora/$numero""").withHeaders("Authorization" -> newToken).get.pipeTo(sender)
    case ReceiveTimeout =>
      val e = TimeOutException("VIGIA: Receive timed out")
      throw e
  }

  override val supervisorStrategy =
    OneForOneStrategy(maxNrOfRetries = 2, withinTimeRange = 1 minute) {
      case _: ArithmeticException      => Resume
      case _: NullPointerException     => Restart
      case _: IllegalArgumentException => Stop
      case _: TimeOutException         => Resume       
      case _: Exception                => Restart
    }
}

对此执行元的调用是验证方法的一部分,如果尝试与ws：

implicit val timeout = Timeout(5 seconds)
lazy val vigiaActor : ActorRef = Akka.system.actorOf(Props[VigiaActor])

(vigiaActor ? VigiaActor.ObraExists(empenho.obra.get, empenho.unidadeGestora)).map {
  case r : WSResponse =>
    val exists = r.body.toBoolean

    if (!exists && empenho.tipoMeta.get.equals(4)) {
      erros.adicionarErro(controle.codigoArquivo, row, line, s"Nº de Obra não informado ou inválido para o Tipo de Meta 4 - Obras" , TipoErroImportacaoEnum.WARNING)
    }

  case _ => erros.adicionarErro(controle.codigoArquivo, row, line, s"Nº de Obra não informado ou inválido para o Tipo de Meta 4 - Obras" , TipoErroImportacaoEnum.WARNING)
}

protected def validateRow(row: Int, line: String, empenho: Empenho, calendarDataEnvioArquivo: Calendar)(implicit s: Session, controle: ControleArquivo, erros:ImportacaoException): Unit = {
    implicit val timeout = Timeout(5 seconds)
    lazy val vigiaActor : ActorRef = Akka.system.actorOf(Props[VigiaActor])

    (vigiaActor ? VigiaActor.ObraExists(empenho.obra.get, empenho.unidadeGestora)).map {
      case e: TimeOutException => println("TIMOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOOUT!!!!")
      case r: WSResponse => {...}
    }
}

case ReceiveTimeout =>
  val e = TimeOutException("VIGIA: Receive timed out")
  sender ! e

与以前一样,我将收到以下日志消息：

[信息][07/20/2017 10:28:05.738][Application-Akka.Actor.Default-Dispatcher-5][Akka://Application/Deadletters]未传递从执行元[Akka://Application/User/$C#1834419855]到执行元[Akka:/Application/Deadletters]的消息[Model.Exception.TimeoutException]。[1]遇到的死信。可以使用配置设置'akka.log-dead-letters'和'akka.log-dead-letters-dider-shutdown'关闭或调整此日志记录。

Answer 1

Context.SetReceiveTimeout(5秒）触发将ReceiveTimeout消息发送到VigiaActor（如果该执行元在5秒内没有收到消息）。Akka在内部将ReceiveTimeout发送给您的执行元，这就是为什么在更新的代码中，尝试将异常发送给Sender并没有达到您预期的效果。换句话说，case ReceiveTimeout=>子句中的sender不是ObraExists消息的原始发送方。

在VigiaActor中设置接收超时与WS请求超时无关，因为如果请求超时，则不会向VigiaActor发送任何消息。即使在ws请求未在5秒内完成时向执行元发送了一条消息，也可能同时在执行元的邮箱中列队了另一条ObraExists消息，因此无法触发ReceiveTimeout。

简而言之，设置执行元的接收超时不是处理ws请求超时的正确机制。（使用当前将Get请求的结果通过管道传送到发件人的方法，您可以调整发件人以处理超时。事实上，我会完全放弃VigiaActor并直接在ValidateRow方法中调用WS。但删除执行元可能不是您问题的重点。）

如果必须在执行元中处理ws请求超时，可以使用如下方法：

import scala.util.{Failure, Success}

class VigiaActor extends akka.actor.Actor {
  import VigiaActor._
  val log = Logging(context.system, this)

  def receive = {
    case ObraExists(numero: String, unidadeGestora: String) =>
      val s = sender // capture the original sender
      WS.url(baseURL + s"""/obras/exists/$unidadeGestora/$numero""")
        .withHeaders("Authorization" -> newToken)
        .withRequestTimeout(5 seconds) // set the timeout
        .get
        .onComplete {
          case Success(resp) =>
            s ! resp
          case Failure(e: scala.concurrent.TimeoutException) =>
            s ! TimeOutException("VIGIA: Receive timed out")
          case Failure(_) =>
            // do something in the case of non-timeout failures
        }
  }
}