AKKA官方文档阅读笔记（3）JAVA版2.5.16

╰半夏微凉° 2022-05-10 03:38 175阅读 0赞

> 以下内容来自官方文档：[https://doc.akka.io/docs/akka/2.5/guide/tutorial\_5.html][https_doc.akka.io_docs_akka_2.5_guide_tutorial_5.html]

## 可能遇到的场景 ##

到目前为止，例子中Actor之间的对话模式都是比较简单的，这个文档里会有个稍微复杂点儿的例子  
接着上个文档的例子，现在有DeviceManager，DeviceGroup，Device三种Actor，我们的目标是查询一个group中所有device的温度信息。  
第一个问题就是，device是动态的，随时可能新增或停止，我们当然可以在查询一开始的时候就约定，我要查这一瞬间存活的device的信息，但是这两会有两个问题：

*  假如一个device在查询过程中停止了，就不会给你响应了
 *  加如一个device在查询过程中新增进来，结果中也不会包含它的信息，因为在查询开始的瞬间没有它

可以这样解决这两个问题：

*  当group收到一个新查询请求时，建立一个当前存活的device的快照，只向这些actor请求数据
 *  之后新增的actor数据将被此次请求忽略
 *  如果快照中的actor在查询期间停止而没有应答，我们将向查询消息的发送者报告这个消息

除了device actor会动态地来来去去之外，一些actor虽然存活，但可能长时间都不应答。 例如，它们可能会陷入意外的死循环中，或者因为代码bug消息处理失败了，或者消息丢了。 我们不希望查询无限期地等待，因此我们将在以下任一情况下认为此次查群请求完成了：

*  快照中的所有actor都已响应或已确认停止。
 *  达到了指定的超时时间。

鉴于这些设定，以及快照中的device可能刚刚启动，还没有接收到温度数据，自然就没有数据可以返回。我们可以针对温度查询为每个设备actor定义四种状态：

*  它启动着并有温度数据： `Temperature`。
 *  它启动着，但还没有温度：`TemperatureNotAvailable`。
 *  它已停止：`DeviceNotAvailable`。
 *  它没有在超时前回复：`DeviceTimedOut`。

让我们把下面地代码添加到`DeviceGroup`类中：

// 查询温度请求 public static final class RequestAllTemperatures { final long requestId; public RequestAllTemperatures(long requestId) { this.requestId = requestId; } } // 查询温度请求响应 public static final class RespondAllTemperatures { final long requestId; final Map<String, TemperatureReading> temperatures; public RespondAllTemperatures(long requestId, Map<String, TemperatureReading> temperatures) { this.requestId = requestId; this.temperatures = temperatures; } } // device返回信息接口 public static interface TemperatureReading { } // 返回温度 public static final class Temperature implements TemperatureReading { public final double value; public Temperature(double value) { this.value = value; } } // 返回暂无数据 public static final class TemperatureNotAvailable implements TemperatureReading { } // 返回device已停止 public static final class DeviceNotAvailable implements TemperatureReading { } // 返回查询超时 public static final class DeviceTimedOut implements TemperatureReading { }

## 执行查询 ##

接下来该干啥呢？一种做法是把对查询请求地处理放在DeviceGroup类中，但是这样子会很麻烦而且容易出错，违背了单一职责原则。所以我们采取更简单更好的方案，我们将创建一个代表单个查询的actor，代替group actor执行完成查询所需的任务。 到目前为止，我们创建的actor都是典型的域对象，但现在，我们将创建一个代表进程或任务而不是实体的actor。 这样，保持我们的device group功能简单，并且能够更好地单独测试查询功能。

**定义查询Actor**  
首先，我们需要设计查询actor的生命周期。 这包括确定其初始状态，初始化操作，结束清理工作(如有必要)。 查询actor将需要以下信息：

*  要查询的device actor的快照和ID。
 *  RequestId（以便我们可以将其包含在回复中）。
 *  查询发起者的actor的引用。 我们会直接将结果发送给这个actor。
 *  超时时间，指明查询应等待回复的时间。 这个参数将简化测试。

**查询超时处理**  
隆重推出调度器：`scheduler`

*  我们从ActorSystem获得调度器，它也可以从actor的上下文访问：`getContext().getSystem().scheduler()`。 参数中有一个`ExecutionContext`，它是执行计时器任务的线程池。 在我们的例子中，我们通过传入`getContext().dispatcher()`来使用与actor相同的调度程序。
 *  `scheduler.scheduleOnce(time,actorRef,message,executor,sender)`方法将在指定时间内将指定消息发送给指定ActorRef。

我们需要一个表示查询超时的消息。 为此，我们创建了一个简单的消息`CollectionTimeout`，没有任何参数。 `scheduleOnce`的返回值是一个可取消的对象(`Cancellable`)，如果查询在指定时间内完成，可用于取消计时器。 在查询开始时，我们需要向device依次询问当前温度。 假如device在返回`ReadTemperature`前就停止了，我们需要收到通知，就要观察(`watch`)每个actor。 这样，actor停止时我们会收到`Terminated`消息，而不需要等到超时才能将这些actor标记为不可用。  
把这些内容整合起来，我们的`DeviceGroupQuery`actor的大致框架如下所示：

**跟踪Actor状态**  
除了挂起的计时器之外，查询actor还有一个有状态切面，就是需要跟踪已经回复，已经停止或没有回复的actor集合。 跟踪此状态的一种方法是在actor中创建可变字段。我们使用另一种方法，利用actor可以修改消息响应方法的能力。 `Receive`只是一个可以从另一个函数返回的函数（或者一个对象，例如python中把函数也当作对象）。 默认情况下，`receive`代码块定义actor处理消息的行为，但可以在actor的生命周期内多次更改它。 我们调用`context.become(newBehavior)`，其中newBehavior的类型为`Receive`。 我们将利用此功能来跟踪actor的状态。  
对于我们的用例：

*  我们不是直接定义receive，而是委托waitingForReplies函数来创建Receive。
 *  waitingForReplies函数将跟踪两个不断变化的值：
    
     *  已经收到的回复
     *  仍在等待的一组actor

我们有三个事件要处理：

*  正常返回温度数据的`RespondTemperature`消息。
 *  actor的终止的`Terminated`消息。
 *  达到超时时间，定时器发送的`CollectionTimeout`。

在前两种情况下，我们需要跟踪回复（维护前面提到的两个集合），我们把这个功能委托给一个方法`receivedResponse`，细节在后面讨论。 在超时的情况下，只需要将`DeviceTimedOut`作为所有仍未恢复的Actor的响应。 然后我们把已经收到响应的结果集发送给查询的发起者并停止查询actor。  
将以下内容添加到DeviceGroupQuery源文件中：

@Override public Receive createReceive() { return waitingForReplies(new HashMap<>(), actorToDeviceId.keySet()); } // 构建Receive需要两个参数，也就是需要维护的两个集合，只要每次改变集合后重新通过此方法获得Receive // 并通过context.become改变自身的消息处理行为，就实现了状态的跟踪 public Receive waitingForReplies( Map<String, DeviceGroup.TemperatureReading> repliesSoFar, Set<ActorRef> stillWaiting) { return receiveBuilder() .match(Device.RespondTemperature.class, r -> { ActorRef deviceActor = getSender(); DeviceGroup.TemperatureReading reading = r.value .map(v -> (DeviceGroup.TemperatureReading) new DeviceGroup.Temperature(v)) .orElse(new DeviceGroup.TemperatureNotAvailable()); receivedResponse(deviceActor, reading, stillWaiting, repliesSoFar); }) .match(Terminated.class, t -> { receivedResponse(t.getActor(), new DeviceGroup.DeviceNotAvailable(), stillWaiting, repliesSoFar); }) .match(CollectionTimeout.class, t -> { Map<String, DeviceGroup.TemperatureReading> replies = new HashMap<>(repliesSoFar); for (ActorRef deviceActor : stillWaiting) { String deviceId = actorToDeviceId.get(deviceActor); replies.put(deviceId, new DeviceGroup.DeviceTimedOut()); } // 将最终结果返回给消息请求者 requester.tell(new DeviceGroup.RespondAllTemperatures(requestId, replies), getSelf()); // 记得要关闭自身 getContext().stop(getSelf()); }) .build(); }

如果看到这里，你还是不太清楚到底是怎么维护`repliesSoFar`结果集和`stillWaiting`结果集的话，你只需要知道：

*  repliesSoFar表示已经得到明确答复的响应集合
 *  stillWaiting表示仍没有回复消息的device集合
 *  waitingForReplies方法本身不处理消息，它返回消息的处理器，或者说，它返回用于处理消息的function
 *  处理消息的function依赖于repliesSoFar和stillWaiting，我们要做的就是在必要的时候，修改这两个集合，传入waitingForReplies以获得消息处理程序（Receive），并使用`context.become(Receive)`来告诉当前actor，用使用最新集合的消息处理程序来处理消息

现在，来弄清楚`receivedResponse`要做什么。 首先，我们需要在`repliesSoFar`中记录得到的响应，并从`stillWaiting`中删除actor。 下一步是检查`stillWaiting`是不是空了。 如果空了，表示所有的device都返回消息了，我们将查询结果发送给原始请求者并停止查询actor。 否则，我们需要使用更新后的repliesSoFar和stillWaiting获得新Receive并等待更多消息。 在之前的代码中，我们将`Terminated`作为隐式响应`DeviceNotAvailable`，因此receivedResponse不需要做任何特殊的事情。 但是，仍有一个小细节我们要考虑到，我们可能会从设备actor收到期望的响应，但随后他有可能就停止了。 我们不希望第二个事件覆盖已经收到的回复。 换句话说，我们不希望在记录响应后收到`Terminated`。 通过调用`context.unwatch(ref)`可以很容易地实现。 多次调用也是安全的，只有第一次调用才会有效，其余的都会被忽略。 有了这些知识，我们可以创建receivedResponse方法：

public void receivedResponse(ActorRef deviceActor, DeviceGroup.TemperatureReading reading, Set<ActorRef> stillWaiting, Map<String, DeviceGroup.TemperatureReading> repliesSoFar) { getContext().unwatch(deviceActor); String deviceId = actorToDeviceId.get(deviceActor); Set<ActorRef> newStillWaiting = new HashSet<>(stillWaiting); newStillWaiting.remove(deviceActor); Map<String, DeviceGroup.TemperatureReading> newRepliesSoFar = new HashMap<>(repliesSoFar); newRepliesSoFar.put(deviceId, reading); if (newStillWaiting.isEmpty()) { requester.tell(new DeviceGroup.RespondAllTemperatures(requestId, newRepliesSoFar), getSelf()); getContext().stop(getSelf()); } else { getContext().become(waitingForReplies(newRepliesSoFar, newStillWaiting)); } }

到了这一刻，你可能很自然地想问，我们使用context.become()而不是使repliesSoFar和stillWaiting成为actor的可变字段？这样做获得了什么， 是的，在这个简单的例子中，好处并不明显。 但是当你突然有更多种类的状态时，这种状态保持的价值变得更加明显。 由于每个状态可能具有相关的临时数据，因此将这些数据作为字段会污染全局状态，即不清楚在什么状态下使用哪些字段。 使用参数化的“工厂”方法，我们可以保持仅与状态相关的数据私有。建议你熟悉我们在此处使用的解决方案，因为它有助于以更清晰，更易于维护的方式构建更复杂的actor代码。  
完整的查询Actor代码如下：

public class DeviceGroupQuery extends AbstractActor { public static final class CollectionTimeout { } private final LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this); final Map<ActorRef, String> actorToDeviceId; final long requestId; final ActorRef requester; Cancellable queryTimeoutTimer; public DeviceGroupQuery(Map<ActorRef, String> actorToDeviceId, long requestId, ActorRef requester, FiniteDuration timeout) { this.actorToDeviceId = actorToDeviceId; this.requestId = requestId; this.requester = requester; queryTimeoutTimer = getContext().getSystem().scheduler().scheduleOnce( timeout, getSelf(), new CollectionTimeout(), getContext().dispatcher(), getSelf() ); } public static Props props(Map<ActorRef, String> actorToDeviceId, long requestId, ActorRef requester, FiniteDuration timeout) { return Props.create(DeviceGroupQuery.class, actorToDeviceId, requestId, requester, timeout); } @Override public void preStart() { for (ActorRef deviceActor : actorToDeviceId.keySet()) { getContext().watch(deviceActor); deviceActor.tell(new Device.ReadTemperature(0L), getSelf()); } } @Override public void postStop() { queryTimeoutTimer.cancel(); } @Override public Receive createReceive() { return waitingForReplies(new HashMap<>(), actorToDeviceId.keySet()); } public Receive waitingForReplies( Map<String, DeviceGroup.TemperatureReading> repliesSoFar, Set<ActorRef> stillWaiting) { return receiveBuilder() .match(Device.RespondTemperature.class, r -> { ActorRef deviceActor = getSender(); DeviceGroup.TemperatureReading reading = r.value .map(v -> (DeviceGroup.TemperatureReading) new DeviceGroup.Temperature(v)) .orElse(new DeviceGroup.TemperatureNotAvailable()); receivedResponse(deviceActor, reading, stillWaiting, repliesSoFar); }) .match(Terminated.class, t -> { receivedResponse(t.getActor(), new DeviceGroup.DeviceNotAvailable(), stillWaiting, repliesSoFar); }) .match(CollectionTimeout.class, t -> { Map<String, DeviceGroup.TemperatureReading> replies = new HashMap<>(repliesSoFar); for (ActorRef deviceActor : stillWaiting) { String deviceId = actorToDeviceId.get(deviceActor); replies.put(deviceId, new DeviceGroup.DeviceTimedOut()); } requester.tell(new DeviceGroup.RespondAllTemperatures(requestId, replies), getSelf()); getContext().stop(getSelf()); }) .build(); } public void receivedResponse(ActorRef deviceActor, DeviceGroup.TemperatureReading reading, Set<ActorRef> stillWaiting, Map<String, DeviceGroup.TemperatureReading> repliesSoFar) { getContext().unwatch(deviceActor); String deviceId = actorToDeviceId.get(deviceActor); Set<ActorRef> newStillWaiting = new HashSet<>(stillWaiting); newStillWaiting.remove(deviceActor); Map<String, DeviceGroup.TemperatureReading> newRepliesSoFar = new HashMap<>(repliesSoFar); newRepliesSoFar.put(deviceId, reading); if (newStillWaiting.isEmpty()) { requester.tell(new DeviceGroup.RespondAllTemperatures(requestId, newRepliesSoFar), getSelf()); getContext().stop(getSelf()); } else { getContext().become(waitingForReplies(newRepliesSoFar, newStillWaiting)); } } }

接下来是很长一段的测试代码，感兴趣的看下原文。

[https_doc.akka.io_docs_akka_2.5_guide_tutorial_5.html]: https://doc.akka.io/docs/akka/2.5/guide/tutorial_5.html