将来如何将错误传播给父actor

Question

我尝试使用akka 和futures 来理解故障处理。 例如，我有父母和孩子演员。

子actor有两种失败案例：
案例 1) 消息处理时发生错误
案例 2) 未来发生错误

在这两种情况下，我都需要将错误传播给父级，但在第二种情况下，它不会发生。我做错了什么？

import akka.actor.SupervisorStrategy.{Decider, Stop}
import akka.actor.{Actor, ActorRef, ActorSystem, OneForOneStrategy, Props, SupervisorStrategy}
import akka.testkit.{TestKit, TestProbe}
import org.junit.{After, Before, Test}

import scala.concurrent.Future
import scala.util.{Failure, Success}


class Parent(_case: String, probe: ActorRef) extends Actor {

  val child = context.actorOf(Props(new Child(_case)), "myLittleChild")

  final val defaultStrategy: SupervisorStrategy = {
    def defaultDecider: Decider = {
      case ex: Exception =>
        probe ! ex
        Stop
    }

    OneForOneStrategy()(defaultDecider)
  }

  override def supervisorStrategy: SupervisorStrategy = defaultStrategy

  override def receive: Receive = {
    case msg => unhandled(msg)
  }

}

class Child(_case: String) extends Actor {

  implicit val ec = context.dispatcher

  override def preStart(): Unit = {
    self ! _case
  }


  override def receive: Receive = {
    case "case1" => throw new RuntimeException("fail")
    case "case2" => Future[String] {
      throw new RuntimeException("fail")
    }.onComplete {
      case Success(s) => println(s)
      case Failure(e) =>
        throw e
    }
    case msg => unhandled(msg)
  }
}


class TestExample {

  protected implicit var system: ActorSystem = _

  @Before
  def setup(): Unit = {
    system = ActorSystem.create("test")
  }

  @After
  def tearDown(): Unit = {
    TestKit.shutdownActorSystem(system)
  }

  @Test
  def case1(): Unit = {
    val testProbe = TestProbe()
    system.actorOf(Props(new Parent("case1", testProbe.ref)))
    testProbe expectMsgClass classOf[RuntimeException]
  }

  @Test
  def case2(): Unit = {
    val testProbe = TestProbe()
    system.actorOf(Props(new Parent("case2", testProbe.ref)))
    testProbe expectMsgClass classOf[RuntimeException]
  }

}

Answer 1

这不是父母与孩子之间交流的方式。 正确的方法是定义一条包含失败的消息（而不是发送异常！）。 然后父母可以适当地处理消息。

此外，在父级中构建子actor 不是首选，因为这使得测试actor 变得非常困难。相反，子actor-factory 函数应该作为参数传递给父actor。然后，在测试 Parent Actor 时，可以很容易地用虚拟 Actor（例如 TestActorRef 或 TestProbe）替换它。同样，可以单独测试子actor，以将正确的消息返回给父actor。

另外，不建议在演员中使用“未来”。 Actor 已经在异步运行，当时只处理 1 条消息。当你开始在actor中使用Future时，你必须处理在Future还没有完成的时候接收到其他消息的情况，因为在Future完成之前，actor可能处于不正确的状态。在演员中使用 Future 的一种方法是使用 book 'Effective Akka'（Extra Pattern，Cameo Pattern）中描述的临时演员。

'Effective Akka' book 是从 Akka 开始的好读物。它包含一些最佳实践和要避免的事情。这是一本小书，读起来很快。

根据评论更新：
在这种情况下，您有 2 个选择：

• 由于actor 已经在运行异步，您可以决定在actor 中使其同步。这将使它更简单，但需要阻塞。

• 其他解决方案是处理 Future 的 onComplete 并向父级（或对结果感兴趣的参与者）发送成功或失败消息。我个人不会抛出异常。

  我会在父actor中传递一个child-actor-factory，对于子actor（或worker）actor，要么传入想要响应的actor，要么从“发送者”获取响应。

  请注意，您必须在调用 Future 之前捕获“发送者”。并且，使用不同的线程池，否则 Future 将使用与 Actor 本身相同的线程池。为了重用这个池，这也是你想要传递给子actor的东西，然后你也可以调整它以进行测试。

我不明白你为什么要重启演员。无论如何，这似乎是一个无状态的参与者，只是一个对 databaseApi 的适配器。

对于子角色的实现，您可以考虑使用额外/客串模式。然后你确定它没有收到其他消息（完成后不要忘记停止演员）。但是，通过将其设置为单独的参与者，您最终可以决定创建一个包含这些参与者的池（使用路由器）来控制并发数据库操作的数量。

Answer 2

要让您的测试通过，您可以将异常发送给参与者并从onComplete 回调之外重新抛出异常：

override def receive: Receive = {
  case "case1" => throw new RuntimeException("fail")
  case "case2" =>
    Future[String] {
      throw new RuntimeException("fail")
    }.onComplete {
      case Success(s) => println(s)
      case Failure(e) =>
        self ! e
    }
  case e: RuntimeException => throw e
  case msg => unhandled(msg)
}

但是，如果您必须在参与者中使用 Future（例如，其方法返回 Future 的第三方库），那么有更好的方法来处理异常。例如，使用您在评论中提到的数据库 API (databaseApi.load(): Future[Rows])，父 Actor 可以向子 Actor 发送 LoadDb 消息，而子 Actor 可以向父 Actor 发送 Rows 或错误消息。孩子的行为如下所示：

def receive = {
  case LoadDb =>
    val s = sender // capture the sender
    databaseApi
      .load
      .onComplete {
        case Success(rows) =>
          s ! rows
        case Failure(e) =>
          s ! DbFailure(e)
      }
  case ...
}

重要的一点是，当孩子收到LoadDb 消息时，我们会制作sender 引用的本地副本，以便从onComplete 回调中获得对正确发件人的引用。如果我们只是在回调中调用sender，这可能会产生错误的结果，因为sender 可能在回调执行时已经改变，正如here 所解释的那样。 （与sender 不同，self 是不可变的，因此在onComplete 中使用self 是安全的。）