容错

依赖

容错（fault tolerance）概念与 Actor 相关，为了使用这些概念，需要在项目中添加如下依赖：

1. <!-- Maven -->
2. <dependency>
3. <groupId>com.typesafe.akka</groupId>
4. <artifactId>akka-actor_2.12</artifactId>
5. <version>2.5.21</version>
6. </dependency>

8. <!-- Gradle -->
9. dependencies {
10. compile group: 'com.typesafe.akka', name: 'akka-actor_2.12', version: '2.5.21'
11. }

13. <!-- sbt -->
14. libraryDependencies += "com.typesafe.akka" %% "akka-actor" % "2.5.21"

简介

正如在「Actor 系统」中所解释的，每个 Actor 都是其子级的监督者，因此每个 Actor 都定义了故障处理的监督策略。这一策略不能在 Actor 系统启动之后改变，因为它是 Actor 系统结构的一个组成部分。

实践中的故障处理

首先，让我们看一个示例，它演示了处理数据存储错误的一种方法，这是现实应用程序中的典型故障源。当然，这取决于实际的应用程序，当数据存储不可用时可以做什么，但是在这个示例中，我们使用了一种尽最大努力的重新连接方法。

阅读以下源代码。内部的注释解释了故障处理的不同部分以及添加它们的原因。强烈建议运行此示例，因为很容易跟踪日志输出以了解运行时发生的情况。

创建监督策略

以下章节将更深入地解释故障处理机制和备选方案。

为了演示，让我们考虑以下策略：

1. private static SupervisorStrategy strategy =
2. new OneForOneStrategy(
3. 10,
4. Duration.ofMinutes(1),
5. DeciderBuilder.match(ArithmeticException.class, e -> SupervisorStrategy.resume())
6. .match(NullPointerException.class, e -> SupervisorStrategy.restart())
7. .match(IllegalArgumentException.class, e -> SupervisorStrategy.stop())
8. .matchAny(o -> SupervisorStrategy.escalate())
9. .build());

11. @Override
12. public SupervisorStrategy supervisorStrategy() {
13. return strategy;
14. }

我们选择了一些众所周知的异常类型，以演示在「监督」中描述的故障处理指令的应用。首先，“一对一策略”意味着每个子级都被单独对待（这和all-for-one策略的效果非常相似，唯一的区别是all-for-one策略中任何决定都适用于监督者的所有子级，而不仅仅是失败的子级）。在上面的示例中，10和Duration.create(1, TimeUnit.MINUTES)分别传递给maxNrOfRetries和withinTimeRange参数，这意味着策略每分钟重新启动一个子级最多10次。如果在withinTimeRange持续时间内重新启动计数超过maxNrOfRetries，则子 Actor 将停止。

此外，这些参数还有一些特殊的值。如果你指定：

  • maxNrOfRetries为-1，withinTimeRange为Duration.Inf()
  • 总是无限制地重新启动子级
  • maxNrOfRetries为-1，withinTimeRange为有限的Duration
  • maxNrOfRetries被视为1
  • maxNrOfRetries为非负数，withinTimeRange为Duration.Inf()
  • withinTimeRange被视为无限持续（即无论需要多长时间），一旦重新启动计数超过maxNrOfRetries，子 Actor 将停止。

构成主体的match语句，由DeciderBuilder的match方法返回的PFBuilder组成，其中builder由build方法完成。这是将子故障类型映射到相应指令的部分。

  • 注释：如果策略在监督者 Actor（而不是单独的类）中声明，则其决策者可以线程安全方式访问 Actor 的所有内部状态，包括获取对当前失败的子级的引用，可用作失败消息的getSender()。

默认监督策略

如果定义的策略不包括引发的异常，则使用升级。

如果没有为 Actor 定义监督策略，则默认情况下会处理以下异常：

  • ActorInitializationException将停止失败的子 Actor
  • ActorKilledException将停止失败的子 Actor
  • DeathPactException将停止失败的子 Actor
  • Exception将重新启动失败的子 Actor
  • 其他类型的Throwable将升级到父级 Actor

如果异常一直升级到根守护者，它将以与上面定义的默认策略相同的方式处理它。

停止监督策略

更接近 Erlang 的方法是在子级失败时采取措施阻止他们，然后在DeathWatch显示子级死亡时由监督者采取纠正措施。此策略还预打包为SupervisorStrategy.stoppingStrategy，并附带一个StoppingSupervisorStrategy配置程序，以便在你希望/user监护者应用它时使用。

记录 Actor 的失败

默认情况下，除非升级，否则SupervisorStrategy会记录故障。升级的故障应该在层次结构中更高的级别处理并记录。

通过在实例化时将loggingEnabled设置为false，可以将SupervisorStrategy的默认日志设置为不可用。定制的日志记录可以在Decider内完成。请注意，当在监督者 Actor 内部声明SupervisorStrategy时，对当前失败的子级的引用可用作sender。

你还可以通过重写logFailure方法自定义自己的SupervisorStrategy中的日志记录。

顶级 Actor 的监督者

顶级 Actor 是指使用system.actorOf()创建的 Actor，它们是「用户守护者」的子代。守护者应用配置的策略，在这种情况下没有应用特殊的规则。

测试应用

下面的部分展示了不同指令在实践中的效果，其中需要一个测试设置。首先，我们需要一个合适的监督者：

1. import akka.japi.pf.DeciderBuilder;
2. import akka.actor.SupervisorStrategy;

4. static class Supervisor extends AbstractActor {

6. private static SupervisorStrategy strategy =
7. new OneForOneStrategy(
8. 10,
9. Duration.ofMinutes(1),
10. DeciderBuilder.match(ArithmeticException.class, e -> SupervisorStrategy.resume())
11. .match(NullPointerException.class, e -> SupervisorStrategy.restart())
12. .match(IllegalArgumentException.class, e -> SupervisorStrategy.stop())
13. .matchAny(o -> SupervisorStrategy.escalate())
14. .build());

16. @Override
17. public SupervisorStrategy supervisorStrategy() {
18. return strategy;
19. }

22. @Override
23. public Receive createReceive() {
24. return receiveBuilder()
25. .match(
26. Props.class,
27. props -> {
28. getSender().tell(getContext().actorOf(props), getSelf());
29. })
30. .build();
31. }
32. }

这个监督者将被用来创建一个子级，我们可以用它进行实验：

1. static class Child extends AbstractActor {
2. int state = 0;

4. @Override
5. public Receive createReceive() {
6. return receiveBuilder()
7. .match(
8. Exception.class,
9. exception -> {
10. throw exception;
11. })
12. .match(Integer.class, i -> state = i)
13. .matchEquals("get", s -> getSender().tell(state, getSelf()))
14. .build();
15. }
16. }

通过使用「TestKit」中描述的实用程序，测试更容易，其中TestProbe提供了一个 Actor 引用，可用于接收和检查回复。

1. import akka.testkit.TestProbe;
2. import akka.testkit.ErrorFilter;
3. import akka.testkit.EventFilter;
4. import akka.testkit.TestEvent;
5. import static java.util.concurrent.TimeUnit.SECONDS;
6. import static akka.japi.Util.immutableSeq;
7. import scala.concurrent.Await;

9. public class FaultHandlingTest extends AbstractJavaTest {
10. static ActorSystem system;
11. scala.concurrent.duration.Duration timeout =
12. scala.concurrent.duration.Duration.create(5, SECONDS);

14. @BeforeClass
15. public static void start() {
16. system = ActorSystem.create("FaultHandlingTest", config);
17. }

19. @AfterClass
20. public static void cleanup() {
21. TestKit.shutdownActorSystem(system);
22. system = null;
23. }

25. @Test
26. public void mustEmploySupervisorStrategy() throws Exception {
27. // code here
28. }
29. }

让我们创建 Actor：

1. Props superprops = Props.create(Supervisor.class);
2. ActorRef supervisor = system.actorOf(superprops, "supervisor");
3. ActorRef child =
4. (ActorRef) Await.result(ask(supervisor, Props.create(Child.class), 5000), timeout);

第一个测试将演示Resume指令，因此我们通过在 Actor 中设置一些非初始状态进行尝试，并使其失败：

1. child.tell(42, ActorRef.noSender());
2. assert Await.result(ask(child, "get", 5000), timeout).equals(42);
3. child.tell(new ArithmeticException(), ActorRef.noSender());
4. assert Await.result(ask(child, "get", 5000), timeout).equals(42);

如你所见，值42保留了故障处理指令。现在，如果我们将失败更改为更严重的NullPointerException，情况将不再如此：

1. child.tell(new NullPointerException(), ActorRef.noSender());
2. assert Await.result(ask(child, "get", 5000), timeout).equals(0);

最后，如果发生致命的IllegalArgumentException，监督者将终止该子级：

1. final TestProbe probe = new TestProbe(system);
2. probe.watch(child);
3. child.tell(new IllegalArgumentException(), ActorRef.noSender());
4. probe.expectMsgClass(Terminated.class);

到目前为止，监督者完全不受子级失败的影响，因为指令集（directives set）确实处理了它。如果出现Exception情况，则情况不再如此，监督者会将失败升级。

1. child = (ActorRef) Await.result(ask(supervisor, Props.create(Child.class), 5000), timeout);
2. probe.watch(child);
3. assert Await.result(ask(child, "get", 5000), timeout).equals(0);
4. child.tell(new Exception(), ActorRef.noSender());
5. probe.expectMsgClass(Terminated.class);

监管者本身由ActorSystem提供的顶级 Actor 进行监督，它具有在所有异常情况下重新启动的默认策略（ActorInitializationException和ActorKilledException的显著异常）。因为重启时的默认指令是杀死所有的子级，所以我们不希望子级在这次失败中幸存。

如果不需要这样做（这取决于用例），我们需要使用一个不同的监督者来覆盖这个行为。

1. static class Supervisor2 extends AbstractActor {

3. private static SupervisorStrategy strategy =
4. new OneForOneStrategy(
5. 10,
6. Duration.ofMinutes(1),
7. DeciderBuilder.match(ArithmeticException.class, e -> SupervisorStrategy.resume())
8. .match(NullPointerException.class, e -> SupervisorStrategy.restart())
9. .match(IllegalArgumentException.class, e -> SupervisorStrategy.stop())
10. .matchAny(o -> SupervisorStrategy.escalate())
11. .build());

13. @Override
14. public SupervisorStrategy supervisorStrategy() {
15. return strategy;
16. }

19. @Override
20. public Receive createReceive() {
21. return receiveBuilder()
22. .match(
23. Props.class,
24. props -> {
25. getSender().tell(getContext().actorOf(props), getSelf());
26. })
27. .build();
28. }

30. @Override
31. public void preRestart(Throwable cause, Optional<Object> msg) {
32. // do not kill all children, which is the default here
33. }
34. }

使用此父级，子级可以在升级的重新启动后存活，如上一个测试所示：

1. superprops = Props.create(Supervisor2.class);
2. supervisor = system.actorOf(superprops);
3. child = (ActorRef) Await.result(ask(supervisor, Props.create(Child.class), 5000), timeout);
4. child.tell(23, ActorRef.noSender());
5. assert Await.result(ask(child, "get", 5000), timeout).equals(23);
6. child.tell(new Exception(), ActorRef.noSender());
7. assert Await.result(ask(child, "get", 5000), timeout).equals(0);

英文原文链接：Fault Tolerance.