如何：常见模式

如何：常见模式

本节列出了常用的 Actor 模式，这些模式被认为是有用的、优雅的或有指导意义的。任何内容都是受欢迎的，例如消息路由策略、监督模式、重启处理等。作为一个特殊的奖励，添加到本节的内容都标记了贡献者的名称，如果在他或她的代码中发现循环模式的每个 Akka 用户都可以为了所有人的利益分享它。在适用的情况下，添加到akka.pattern包中以创建「类似于 OTP 库」也是有意义的。

调度周期消息

详见「Actor Timers」。

具有高级错误报告的一次性 Actor 树

从 Java 进入 Actor 世界的一个好方法是使用Patterns.ask()。这个方法启动一个临时 Actor 来转发消息，并从 Actor 那里收集要“询问”的结果。如果所请求的 Actor 中出现错误，则将接管默认的监督处理。Patterns.ask()的调用方将不会收到通知。

如果调用者对这种异常感兴趣，他们必须确保被询问的 Actor 以Status.Failure(Throwable)进行答复。在被询问的 Actor 之后，可能会生成一个复杂的 Actor 层次结构来完成异步工作。然后监督是控制错误处理的既定方法。

不幸的是，被问到的 Actor 必须了解监督，并且必须捕获异常。这样的 Actor 不太可能在不同的 Actor 层次结构中重用，并且包含残缺的try/catch块。

此模式提供了一种将监督和错误传播封装到临时 Actor 的方法。最后，由Patterns.ask()返回的承诺作为一个失败来实现，包括异常，详见「 Java 8 兼容性」。

让我们看一下示例代码：

1. /*
2. * Copyright (C) 2018-2019 Lightbend Inc. <https://www.lightbend.com>
3. */

5. package jdocs.pattern;

7. import java.util.concurrent.CompletionStage;
8. import java.util.concurrent.TimeoutException;
9. import java.time.Duration;

11. import akka.actor.ActorKilledException;
12. import akka.actor.ActorRef;
13. import akka.actor.ActorRefFactory;
14. import akka.actor.Cancellable;
15. import akka.actor.OneForOneStrategy;
16. import akka.actor.Props;
17. import akka.actor.Scheduler;
18. import akka.actor.Status;
19. import akka.actor.SupervisorStrategy;
20. import akka.actor.Terminated;
21. import akka.actor.AbstractActor;
22. import akka.pattern.Patterns;

24. public class SupervisedAsk {

26. private static class AskParam {
27. Props props;
28. Object message;
29. Duration timeout;

31. AskParam(Props props, Object message, Duration timeout) {
32. this.props = props;
33. this.message = message;
34. this.timeout = timeout;
35. }
36. }

38. private static class AskTimeout {}

40. public static class AskSupervisorCreator extends AbstractActor {

42. @Override
43. public Receive createReceive() {
44. return receiveBuilder()
45. .match(
46. AskParam.class,
47. message -> {
48. ActorRef supervisor = getContext().actorOf(Props.create(AskSupervisor.class));
49. supervisor.forward(message, getContext());
50. })
51. .build();
52. }
53. }

55. public static class AskSupervisor extends AbstractActor {
56. private ActorRef targetActor;
57. private ActorRef caller;
58. private AskParam askParam;
59. private Cancellable timeoutMessage;

61. @Override
62. public SupervisorStrategy supervisorStrategy() {
63. return new OneForOneStrategy(
64. 0,
65. Duration.ZERO,
66. cause -> {
67. caller.tell(new Status.Failure(cause), getSelf());
68. return SupervisorStrategy.stop();
69. });
70. }

72. @Override
73. public Receive createReceive() {
74. return receiveBuilder()
75. .match(
76. AskParam.class,
77. message -> {
78. askParam = message;
79. caller = getSender();
80. targetActor = getContext().actorOf(askParam.props);
81. getContext().watch(targetActor);
82. targetActor.forward(askParam.message, getContext());
83. Scheduler scheduler = getContext().getSystem().scheduler();
84. timeoutMessage =
85. scheduler.scheduleOnce(
86. askParam.timeout,
87. getSelf(),
88. new AskTimeout(),
89. getContext().getDispatcher(),
90. null);
91. })
92. .match(
93. Terminated.class,
94. message -> {
95. Throwable ex = new ActorKilledException("Target actor terminated.");
96. caller.tell(new Status.Failure(ex), getSelf());
97. timeoutMessage.cancel();
98. getContext().stop(getSelf());
99. })
100. .match(
101. AskTimeout.class,
102. message -> {
103. Throwable ex =
104. new TimeoutException(
105. "Target actor timed out after " + askParam.timeout.toString());
106. caller.tell(new Status.Failure(ex), getSelf());
107. getContext().stop(getSelf());
108. })
109. .build();
110. }
111. }

113. public static CompletionStage<Object> askOf(
114. ActorRef supervisorCreator, Props props, Object message, Duration timeout) {
115. AskParam param = new AskParam(props, message, timeout);
116. return Patterns.ask(supervisorCreator, param, timeout);
117. }

119. public static synchronized ActorRef createSupervisorCreator(ActorRefFactory factory) {
120. return factory.actorOf(Props.create(AskSupervisorCreator.class));
121. }
122. }

在askOf方法中，会向SupervisorCreator发送用户消息。SupervisorCreator创建一个SupervisorActor并转发消息。这可以防止由于 Actor 创建而导致 Actor 系统过载。监督者负责创建用户 Actor、转发消息、处理 Actor 终止和监督。此外，如果执行时间过期，则监管者将停止用户 Actor。

如果发生异常，监督者会通知临时 Actor 引发了哪个异常。之后，Actor 层次结构停止。

最后，我们能够执行一个 Actor 并接收结果或异常。

1. /*
2. * Copyright (C) 2018-2019 Lightbend Inc. <https://www.lightbend.com>
3. */

5. package jdocs.pattern;

7. import akka.actor.ActorRef;
8. import akka.actor.ActorRefFactory;
9. import akka.actor.Props;
10. import akka.actor.AbstractActor;
11. import akka.util.Timeout;
12. import scala.concurrent.duration.FiniteDuration;

14. import java.time.Duration;
15. import java.util.concurrent.CompletionStage;
16. import java.util.concurrent.TimeUnit;

18. public class SupervisedAskSpec {

20. public Object execute(
21. Class<? extends AbstractActor> someActor,
22. Object message,
23. Duration timeout,
24. ActorRefFactory actorSystem)
25. throws Exception {
26. // example usage
27. try {
28. ActorRef supervisorCreator = SupervisedAsk.createSupervisorCreator(actorSystem);
29. CompletionStage<Object> finished =
30. SupervisedAsk.askOf(supervisorCreator, Props.create(someActor), message, timeout);
31. return finished.toCompletableFuture().get(timeout.toMillis(), TimeUnit.MILLISECONDS);
32. } catch (Exception e) {
33. // exception propagated by supervision
34. throw e;
35. }
36. }
37. }

可扩展的分布式事件源和 CQRS

「Lagom 框架」编码了将 Akka 持久性和 Akka 持久性查询与集群分片相结合的许多最佳实践，以构建具有事件源和 CQRS 的可扩展和弹性系统。

请参见 Lagom 文档中的「管理数据持久性」和「持久性实体」。

英文原文链接：HowTo: Common Patterns.