第 5 部分: 查询设备组

依赖

在你的项目中添加如下依赖：

  1. <!-- Maven -->
  2. <dependency>
  3.   <groupId>com.typesafe.akka</groupId>
  4.   <artifactId>akka-actor_2.11</artifactId>
  5.   <version>2.5.19</version>
  6. </dependency>
  8. <!-- Gradle -->
  9. dependencies {
  10.   compile group: 'com.typesafe.akka', name: 'akka-actor_2.11', version: '2.5.19'
  11. }
  13. <!-- sbt -->
  14. libraryDependencies += "com.typesafe.akka" %% "akka-actor" % "2.5.19"

简介

到目前为止，我们所看到的对话模式很简单，因为它们要求 Actor 保持很少或根本就没有状态。明确地：

  • 设备 Actor 返回一个不需要状态更改的读取
  • 记录温度，更新单个字段
  • 设备组 Actor 通过添加或删除映射中的条目来维护组成员身份

在本部分中，我们将使用一个更复杂的示例。由于房主会对整个家庭的温度感兴趣，我们的目标是能够查询一个组中的所有设备 Actor。让我们先研究一下这样的查询 API 应该如何工作。

处理可能的情况

我们面临的第一个问题是，一个组的成员是动态的。每个传感器设备都由一个可以随时停止的 Actor 表示。在查询开始时，我们可以询问所有现有设备 Actor 当前的温度。但是，在查询的生命周期中：

  • 设备 Actor 可能会停止工作，无法用温度读数做出响应。
  • 一个新的设备 Actor 可能会启动，并且不会包含在查询中，因为我们不知道它。

这些问题可以用许多不同的方式来解决，但重要的是要解决所期望的行为。以下工作对于我们的用例是很有用的：

  • 当查询到达时，组 Actor 将获取现有设备 Actor 的快照（snapshot），并且只向这些 Actor 询问温度。
  • 查询到达后启动的 Actor 可以被忽略。
  • 如果快照中的某个 Actor 在查询期间停止而没有应答，我们将向查询消息的发送者报告它停止的事实。

除了设备 Actor 动态地变化之外，一些 Actor 可能需要很长时间来响应。例如，它们可能被困在一个意外的无限循环中，或者由于一个 bug 而失败，并放弃我们的请求。我们不希望查询无限期地继续，因此在以下任何一种情况下，我们都会认为它是完成的：

  • 快照中的所有 Actor 要么已响应，要么确认已停止。
  • 我们达到了预定的（pre-defined）最后期限。

考虑到这些决定，再加上快照中的设备可能刚刚启动但尚未接收到要记录的温度，我们可以针对温度查询为每个设备 Actor 定义四种状态：

  • 它有一个可用的温度：Temperature。
  • 它已经响应，但还没有可用的温度：TemperatureNotAvailable。
  • 它在响应之前已停止：DeviceNotAvailable。
  • 它在最后期限之前没有响应：DeviceTimedOut。

在消息类型中汇总这些信息，我们可以将以下代码添加到DeviceGroup：

  1. public static final class RequestAllTemperatures {
  2.   final long requestId;
  4.   public RequestAllTemperatures(long requestId) {
  5.     this.requestId = requestId;
  6.   }
  7. }
  9. public static final class RespondAllTemperatures {
  10.   final long requestId;
  11.   final Map<String, TemperatureReading> temperatures;
  13.   public RespondAllTemperatures(long requestId, Map<String, TemperatureReading> temperatures) {
  14.     this.requestId = requestId;
  15.     this.temperatures = temperatures;
  16.   }
  17. }
  19. public static interface TemperatureReading {
  20. }
  22. public static final class Temperature implements TemperatureReading {
  23.   public final double value;
  25.   public Temperature(double value) {
  26.     this.value = value;
  27.   }
  29.   @Override
  30.   public boolean equals(Object o) {
  31.     if (this == o) return true;
  32.     if (o == null || getClass() != o.getClass()) return false;
  34.     Temperature that = (Temperature) o;
  36.     return Double.compare(that.value, value) == 0;
  37.   }
  39.   @Override
  40.   public int hashCode() {
  41.     long temp = Double.doubleToLongBits(value);
  42.     return (int) (temp ^ (temp >>> 32));
  43.   }
  45.   @Override
  46.   public String toString() {
  47.     return "Temperature{" +
  48.       "value=" + value +
  49.       '}';
  50.   }
  51. }
  53. public enum TemperatureNotAvailable implements TemperatureReading {
  54.   INSTANCE
  55. }
  57. public enum DeviceNotAvailable implements TemperatureReading {
  58.   INSTANCE
  59. }
  61. public enum DeviceTimedOut implements TemperatureReading {
  62.   INSTANCE
  63. }

实现查询功能

实现查询的一种方法是向组设备 Actor 添加代码。然而，在实践中，这可能非常麻烦并且容易出错。请记住，当我们启动查询时，我们需要获取当前设备的快照并启动计时器，以便强制执行截止时间。同时，另一个查询可以到达。对于第二个查询，我们需要跟踪完全相同的信息，但与前一个查询隔离。这将要求我们在查询和设备 Actor 之间维护单独的映射。

相反，我们将实现一种更简单、更优雅的方法。我们将创建一个表示单个查询的 Actor，并代表组 Actor 执行完成查询所需的任务。到目前为止，我们已经创建了属于典型域对象（classical domain）的 Actor，但是现在，我们将创建一个表示流程或任务而不是实体的 Actor。我们通过保持我们的组设备 Actor 简单和能够更好地隔离测试查询功能而受益。

定义查询 Actor

首先，我们需要设计查询 Actor 的生命周期。这包括识别其初始状态、将要采取的第一个操作以及清除（如果需要）。查询 Actor 需要以下信息：

  • 要查询的活动设备 Actor 的快照和 ID。
  • 启动查询的请求的 ID（以便我们可以在响应中包含它）。
  • 发送查询的 Actor 的引用。我们会直接给这个 Actor 响应。
  • 指示查询等待响应的期限。将其作为参数将简化测试。

设置查询超时

由于我们需要一种方法来指示我们愿意等待响应的时间，现在是时候引入一个我们还没有使用的新的 Akka 特性，即内置的调度器（built-in scheduler）功能了。使用调度器（scheduler）很简单：

  • 我们可以从ActorSystem中获取调度器，而ActorSystem又可以从 Actor 的上下文中访问：getContext().getSystem().scheduler()。这需要一个ExecutionContext，它是将执行计时器任务本身的线程池。在我们的示例中，我们通过传入getContext().dispatcher()来使用与 Actor 相同的调度器。
  • scheduler.scheduleOnce(time, actorRef, message, executor, sender)方法将在指定的time将消息message调度到Future，并将其发送给 Actor 的ActorRef。

我们需要创建一个表示查询超时的消息。为此，我们创建了一个没有任何参数的简单消息CollectionTimeout。scheduleOnce的返回值是Cancellable，如果查询及时成功完成，可以使用它取消定时器。在查询开始时，我们需要询问每个设备 Actor 当前的温度。为了能够快速检测那些在ReadTemperature信息之前停止的设备，我们还将观察每个 Actor。这样，对于那些在查询生命周期中停止的消息，我们就可以得到Terminated消息，因此我们不需要等到超时时再将这些消息标记为不可用。

综上所述，DeviceGroupQuery Actor 的代码大致如下：

  1. public class DeviceGroupQuery extends AbstractActor {
  2.   public static final class CollectionTimeout {
  3.   }
  5.   private final LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);
  7.   final Map<ActorRef, String> actorToDeviceId;
  8.   final long requestId;
  9.   final ActorRef requester;
  11.   Cancellable queryTimeoutTimer;
  13.   public DeviceGroupQuery(Map<ActorRef, String> actorToDeviceId, long requestId, ActorRef requester, FiniteDuration timeout) {
  14.     this.actorToDeviceId = actorToDeviceId;
  15.     this.requestId = requestId;
  16.     this.requester = requester;
  18.     queryTimeoutTimer = getContext().getSystem().scheduler().scheduleOnce(
  19.             timeout, getSelf(), new CollectionTimeout(), getContext().dispatcher(), getSelf()
  20.     );
  21.   }
  23.   public static Props props(Map<ActorRef, String> actorToDeviceId, long requestId, ActorRef requester, FiniteDuration timeout) {
  24.     return Props.create(DeviceGroupQuery.class, () -> new DeviceGroupQuery(actorToDeviceId, requestId, requester, timeout));
  25.   }
  27.   @Override
  28.   public void preStart() {
  29.     for (ActorRef deviceActor : actorToDeviceId.keySet()) {
  30.       getContext().watch(deviceActor);
  31.       deviceActor.tell(new Device.ReadTemperature(0L), getSelf());
  32.     }
  33.   }
  35.   @Override
  36.   public void postStop() {
  37.     queryTimeoutTimer.cancel();
  38.   }
  39. }

跟踪 Actor 状态

除了挂起的定时器之外，查询 Actor 还有一个状态方面，它跟踪一组 Actor：已回复、已停止或未回复。跟踪此状态的一种方法是在 Actor 中创建可变字段。另一种方法利用改变 Actor 对消息的响应方式的能力。Receive是一个可以从另一个函数返回的函数（如果你愿意的话，也可以是对象）。默认情况下，receive块定义了 Actor 的行为，但在 Actor 的生命周期中可以多次更改它。我们调用context.become(newBehavior)，其中newBehavior是任何类型的Receive。我们将利用此功能跟踪 Actor 的状态。

对于我们的用例：

  • 我们不直接定义receive，而是委托waitingForReplies函数来创建Receive。
  • waitingForReplies函数将跟踪两个更改的值：
    • 已收到响应的Map；
    • 我们还在等待 Actor 响应的Set。

我们有三件事要做：

  • 我们可以从其中一个设备接收RespondTemperature。
  • 我们可以为同时被停止的设备 Actor 接收Terminated的消息。
  • 我们可以达到截止时间（deadline）并收到一个CollectionTimeout消息。

在前两种情况下，我们需要跟踪响应，现在我们将其委托给receivedResponse方法，稍后我们将讨论该方法。在超时的情况下，我们需要简单地把所有还没有响应的 Actor（集合stillWaiting的成员）放在DeviceTimedOut中作为最终响应的状态。然后我们用收集到的结果回复查询提交者，并停止查询 Actor。

要完成此操作，请将以下代码添加到DeviceGroupQuery源文件中：

  1. @Override
  2. public Receive createReceive() {
  3.   return waitingForReplies(new HashMap<>(), actorToDeviceId.keySet());
  4. }
  6. public Receive waitingForReplies(
  7.         Map<String, DeviceGroup.TemperatureReading> repliesSoFar,
  8.         Set<ActorRef> stillWaiting) {
  9.   return receiveBuilder()
  10.           .match(Device.RespondTemperature.class, r -> {
  11.             ActorRef deviceActor = getSender();
  12.             DeviceGroup.TemperatureReading reading = r.value
  13.                     .map(v -> (DeviceGroup.TemperatureReading) new DeviceGroup.Temperature(v))
  14.                     .orElse(DeviceGroup.TemperatureNotAvailable.INSTANCE);
  15.             receivedResponse(deviceActor, reading, stillWaiting, repliesSoFar);
  16.           })
  17.           .match(Terminated.class, t -> {
  18.             receivedResponse(t.getActor(), DeviceGroup.DeviceNotAvailable.INSTANCE, stillWaiting, repliesSoFar);
  19.           })
  20.           .match(CollectionTimeout.class, t -> {
  21.             Map<String, DeviceGroup.TemperatureReading> replies = new HashMap<>(repliesSoFar);
  22.             for (ActorRef deviceActor : stillWaiting) {
  23.               String deviceId = actorToDeviceId.get(deviceActor);
  24.               replies.put(deviceId, DeviceGroup.DeviceTimedOut.INSTANCE);
  25.             }
  26.             requester.tell(new DeviceGroup.RespondAllTemperatures(requestId, replies), getSelf());
  27.             getContext().stop(getSelf());
  28.           })
  29.           .build();
  30. }

目前还不清楚我们将如何“改变”repliesSoFar和stillWaiting数据结构。需要注意的一点是，waitingForReplies函数不能直接处理消息。它返回一个Receive函数来处理消息。这意味着，如果我们使用不同的参数再次调用waitingForReplies，那么它将返回一个全新的Receive，该Receive将使用这些新参数。

我们已经看到了如何通过从receive的返回来安装（install）初始化Receive。例如，为了安装一个新的Receive，为了记录一个新的回复，我们需要一些机制。此机制是方法context.become(newReceive)，它将 Actor 的消息处理函数更改为提供的newReceive函数。可以想象，在开始之前，Actor 会自动调用context.become(receive)，即安装从receive返回的Receive函数。这是另一个重要的观察：处理消息的不是receive，而是返回一个实际处理消息的Receive函数。

我们现在必须弄清楚在receivedResponse中该怎么做。首先，我们需要在repliesSoFar中记录新的结果，并将 Actor 从stillWaiting中移除。下一步是检查是否还有其他我们正在等待的 Actor。如果没有，我们将查询结果发送给原始请求者并停止查询 Actor。否则，我们需要更新repliesSoFar和stillWaiting结构并等待更多的消息。

在之前的代码中，我们将Terminated视为隐式响应DeviceNotAvailable，因此receivedResponse不需要执行任何特殊操作。但是，还有一个小任务我们仍然需要做。我们可能从设备 Actor 那里接收到正确的响应，但是在查询的生命周期中，它会停止。我们不希望此第二个事件覆盖已收到的响应。换句话说，我们不希望在记录响应之后接收Terminated。这很容易通过调用context.unwatch(ref)实现。此方法还确保我们不会接收已经在 Actor 邮箱中的Terminated事件。多次调用此函数也是安全的，只有第一次调用才会有任何效果，其余的调用将被忽略

通过以上的分析，我们创建receivedResponse方法为：

  1. public void receivedResponse(ActorRef deviceActor,
  2.                              DeviceGroup.TemperatureReading reading,
  3.                              Set<ActorRef> stillWaiting,
  4.                              Map<String, DeviceGroup.TemperatureReading> repliesSoFar) {
  5.   getContext().unwatch(deviceActor);
  6.   String deviceId = actorToDeviceId.get(deviceActor);
  8.   Set<ActorRef> newStillWaiting = new HashSet<>(stillWaiting);
  9.   newStillWaiting.remove(deviceActor);
  11.   Map<String, DeviceGroup.TemperatureReading> newRepliesSoFar = new HashMap<>(repliesSoFar);
  12.   newRepliesSoFar.put(deviceId, reading);
  13.   if (newStillWaiting.isEmpty()) {
  14.     requester.tell(new DeviceGroup.RespondAllTemperatures(requestId, newRepliesSoFar), getSelf());
  15.     getContext().stop(getSelf());
  16.   } else {
  17.     getContext().become(waitingForReplies(newRepliesSoFar, newStillWaiting));
  18.   }
  19. }

在这一点上，我们很自然地会问，使用context.become()技巧，而不是使repliesSoFar和stillWaiting结构成为 Actor 的可变字段（例如，vars），我们获得了什么？在这个简单的例子中，没有那么多。当你突然有更多的状态时，这种状态保持的价值变得更加明显。由于每个状态可能都有与其自身相关的临时数据，因此将这些数据作为字段保存会污染 Actor 的全局状态，也就是说，不清楚在什么状态下使用了哪些字段。使用参数化的Receive“工厂”方法，我们可以保持仅与状态相关的数据私有化。使用可变字段而不是context.become()重写查询仍然是一个很好的练习。但是，建议你熟悉我们在这里使用的解决方案，因为它有助于以更干净和更可维护的方式构造更复杂的 Actor 代码。

现在，我们的查询 Actor 完成了，代码如下：

  1. public class DeviceGroupQuery extends AbstractActor {
  2.   public static final class CollectionTimeout {
  3.   }
  5.   private final LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);
  7.   final Map<ActorRef, String> actorToDeviceId;
  8.   final long requestId;
  9.   final ActorRef requester;
  11.   Cancellable queryTimeoutTimer;
  13.   public DeviceGroupQuery(Map<ActorRef, String> actorToDeviceId, long requestId, ActorRef requester, FiniteDuration timeout) {
  14.     this.actorToDeviceId = actorToDeviceId;
  15.     this.requestId = requestId;
  16.     this.requester = requester;
  18.     queryTimeoutTimer = getContext().getSystem().scheduler().scheduleOnce(
  19.             timeout, getSelf(), new CollectionTimeout(), getContext().dispatcher(), getSelf()
  20.     );
  21.   }
  23.   public static Props props(Map<ActorRef, String> actorToDeviceId, long requestId, ActorRef requester, FiniteDuration timeout) {
  24.     return Props.create(DeviceGroupQuery.class, () -> new DeviceGroupQuery(actorToDeviceId, requestId, requester, timeout));
  25.   }
  27.   @Override
  28.   public void preStart() {
  29.     for (ActorRef deviceActor : actorToDeviceId.keySet()) {
  30.       getContext().watch(deviceActor);
  31.       deviceActor.tell(new Device.ReadTemperature(0L), getSelf());
  32.     }
  33.   }
  35.   @Override
  36.   public void postStop() {
  37.     queryTimeoutTimer.cancel();
  38.   }
  40.   @Override
  41.   public Receive createReceive() {
  42.     return waitingForReplies(new HashMap<>(), actorToDeviceId.keySet());
  43.   }
  45.   public Receive waitingForReplies(
  46.           Map<String, DeviceGroup.TemperatureReading> repliesSoFar,
  47.           Set<ActorRef> stillWaiting) {
  48.     return receiveBuilder()
  49.             .match(Device.RespondTemperature.class, r -> {
  50.               ActorRef deviceActor = getSender();
  51.               DeviceGroup.TemperatureReading reading = r.value
  52.                       .map(v -> (DeviceGroup.TemperatureReading) new DeviceGroup.Temperature(v))
  53.                       .orElse(DeviceGroup.TemperatureNotAvailable.INSTANCE);
  54.               receivedResponse(deviceActor, reading, stillWaiting, repliesSoFar);
  55.             })
  56.             .match(Terminated.class, t -> {
  57.               receivedResponse(t.getActor(), DeviceGroup.DeviceNotAvailable.INSTANCE, stillWaiting, repliesSoFar);
  58.             })
  59.             .match(CollectionTimeout.class, t -> {
  60.               Map<String, DeviceGroup.TemperatureReading> replies = new HashMap<>(repliesSoFar);
  61.               for (ActorRef deviceActor : stillWaiting) {
  62.                 String deviceId = actorToDeviceId.get(deviceActor);
  63.                 replies.put(deviceId, DeviceGroup.DeviceTimedOut.INSTANCE);
  64.               }
  65.               requester.tell(new DeviceGroup.RespondAllTemperatures(requestId, replies), getSelf());
  66.               getContext().stop(getSelf());
  67.             })
  68.             .build();
  69.   }
  71.   public void receivedResponse(ActorRef deviceActor,
  72.                                DeviceGroup.TemperatureReading reading,
  73.                                Set<ActorRef> stillWaiting,
  74.                                Map<String, DeviceGroup.TemperatureReading> repliesSoFar) {
  75.     getContext().unwatch(deviceActor);
  76.     String deviceId = actorToDeviceId.get(deviceActor);
  78.     Set<ActorRef> newStillWaiting = new HashSet<>(stillWaiting);
  79.     newStillWaiting.remove(deviceActor);
  81.     Map<String, DeviceGroup.TemperatureReading> newRepliesSoFar = new HashMap<>(repliesSoFar);
  82.     newRepliesSoFar.put(deviceId, reading);
  83.     if (newStillWaiting.isEmpty()) {
  84.       requester.tell(new DeviceGroup.RespondAllTemperatures(requestId, newRepliesSoFar), getSelf());
  85.       getContext().stop(getSelf());
  86.     } else {
  87.       getContext().become(waitingForReplies(newRepliesSoFar, newStillWaiting));
  88.     }
  89.   }
  90. }

测试查询 Actor

现在，让我们验证查询 Actor 实现的正确性。我们需要单独测试各种场景，以确保一切都按预期工作。为了能够做到这一点，我们需要以某种方式模拟设备 Actor 来运行各种正常或故障场景。幸运的是，我们将合作者（collaborators）列表（实际上是一个Map）作为查询 Actor 的参数，这样我们就可以传入TestKit引用。在我们的第一个测试中，我们在有两个设备的情况下进行测试，两个设备都报告了温度：

  1. @Test
  2. public void testReturnTemperatureValueForWorkingDevices() {
  3.   TestKit requester = new TestKit(system);
  5.   TestKit device1 = new TestKit(system);
  6.   TestKit device2 = new TestKit(system);
  8.   Map<ActorRef, String> actorToDeviceId = new HashMap<>();
  9.   actorToDeviceId.put(device1.getRef(), "device1");
  10.   actorToDeviceId.put(device2.getRef(), "device2");
  12.   ActorRef queryActor = system.actorOf(DeviceGroupQuery.props(
  13.           actorToDeviceId,
  14.           1L,
  15.           requester.getRef(),
  16.           new FiniteDuration(3, TimeUnit.SECONDS)));
  18.   assertEquals(0L, device1.expectMsgClass(Device.ReadTemperature.class).requestId);
  19.   assertEquals(0L, device2.expectMsgClass(Device.ReadTemperature.class).requestId);
  21.   queryActor.tell(new Device.RespondTemperature(0L, Optional.of(1.0)), device1.getRef());
  22.   queryActor.tell(new Device.RespondTemperature(0L, Optional.of(2.0)), device2.getRef());
  24.   DeviceGroup.RespondAllTemperatures response = requester.expectMsgClass(DeviceGroup.RespondAllTemperatures.class);
  25.   assertEquals(1L, response.requestId);
  27.   Map<String, DeviceGroup.TemperatureReading> expectedTemperatures = new HashMap<>();
  28.   expectedTemperatures.put("device1", new DeviceGroup.Temperature(1.0));
  29.   expectedTemperatures.put("device2", new DeviceGroup.Temperature(2.0));
  31.   assertEquals(expectedTemperatures, response.temperatures);
  32. }

这是一个很好的例子，但我们知道有时设备不能提供温度测量。这种情况与前一种情况略有不同：

  1. @Test
  2. public void testReturnTemperatureNotAvailableForDevicesWithNoReadings() {
  3.   TestKit requester = new TestKit(system);
  5.   TestKit device1 = new TestKit(system);
  6.   TestKit device2 = new TestKit(system);
  8.   Map<ActorRef, String> actorToDeviceId = new HashMap<>();
  9.   actorToDeviceId.put(device1.getRef(), "device1");
  10.   actorToDeviceId.put(device2.getRef(), "device2");
  12.   ActorRef queryActor = system.actorOf(DeviceGroupQuery.props(
  13.           actorToDeviceId,
  14.           1L,
  15.           requester.getRef(),
  16.           new FiniteDuration(3, TimeUnit.SECONDS)));
  18.   assertEquals(0L, device1.expectMsgClass(Device.ReadTemperature.class).requestId);
  19.   assertEquals(0L, device2.expectMsgClass(Device.ReadTemperature.class).requestId);
  21.   queryActor.tell(new Device.RespondTemperature(0L, Optional.empty()), device1.getRef());
  22.   queryActor.tell(new Device.RespondTemperature(0L, Optional.of(2.0)), device2.getRef());
  24.   DeviceGroup.RespondAllTemperatures response = requester.expectMsgClass(DeviceGroup.RespondAllTemperatures.class);
  25.   assertEquals(1L, response.requestId);
  27.   Map<String, DeviceGroup.TemperatureReading> expectedTemperatures = new HashMap<>();
  28.   expectedTemperatures.put("device1", DeviceGroup.TemperatureNotAvailable.INSTANCE);
  29.   expectedTemperatures.put("device2", new DeviceGroup.Temperature(2.0));
  31.   assertEquals(expectedTemperatures, response.temperatures);
  32. }

我们也知道，有时设备 Actor 会在响应之前停止：

  1. @Test
  2. public void testReturnDeviceNotAvailableIfDeviceStopsBeforeAnswering() {
  3.   TestKit requester = new TestKit(system);
  5.   TestKit device1 = new TestKit(system);
  6.   TestKit device2 = new TestKit(system);
  8.   Map<ActorRef, String> actorToDeviceId = new HashMap<>();
  9.   actorToDeviceId.put(device1.getRef(), "device1");
  10.   actorToDeviceId.put(device2.getRef(), "device2");
  12.   ActorRef queryActor = system.actorOf(DeviceGroupQuery.props(
  13.           actorToDeviceId,
  14.           1L,
  15.           requester.getRef(),
  16.           new FiniteDuration(3, TimeUnit.SECONDS)));
  18.   assertEquals(0L, device1.expectMsgClass(Device.ReadTemperature.class).requestId);
  19.   assertEquals(0L, device2.expectMsgClass(Device.ReadTemperature.class).requestId);
  21.   queryActor.tell(new Device.RespondTemperature(0L, Optional.of(1.0)), device1.getRef());
  22.   device2.getRef().tell(PoisonPill.getInstance(), ActorRef.noSender());
  24.   DeviceGroup.RespondAllTemperatures response = requester.expectMsgClass(DeviceGroup.RespondAllTemperatures.class);
  25.   assertEquals(1L, response.requestId);
  27.   Map<String, DeviceGroup.TemperatureReading> expectedTemperatures = new HashMap<>();
  28.   expectedTemperatures.put("device1", new DeviceGroup.Temperature(1.0));
  29.   expectedTemperatures.put("device2", DeviceGroup.DeviceNotAvailable.INSTANCE);
  31.   assertEquals(expectedTemperatures, response.temperatures);
  32. }

如果你还记得，还有一个用例与设备 Actor 停止相关。我们可以从一个设备 Actor 得到一个正常的回复，但是随后接收到同一个 Actor 的一个Terminated消息。在这种情况下，我们希望保持第一次回复，而不是将设备标记为DeviceNotAvailable。我们也应该测试一下：

  1. @Test
  2. public void testReturnTemperatureReadingEvenIfDeviceStopsAfterAnswering() {
  3.   TestKit requester = new TestKit(system);
  5.   TestKit device1 = new TestKit(system);
  6.   TestKit device2 = new TestKit(system);
  8.   Map<ActorRef, String> actorToDeviceId = new HashMap<>();
  9.   actorToDeviceId.put(device1.getRef(), "device1");
  10.   actorToDeviceId.put(device2.getRef(), "device2");
  12.   ActorRef queryActor = system.actorOf(DeviceGroupQuery.props(
  13.           actorToDeviceId,
  14.           1L,
  15.           requester.getRef(),
  16.           new FiniteDuration(3, TimeUnit.SECONDS)));
  18.   assertEquals(0L, device1.expectMsgClass(Device.ReadTemperature.class).requestId);
  19.   assertEquals(0L, device2.expectMsgClass(Device.ReadTemperature.class).requestId);
  21.   queryActor.tell(new Device.RespondTemperature(0L, Optional.of(1.0)), device1.getRef());
  22.   queryActor.tell(new Device.RespondTemperature(0L, Optional.of(2.0)), device2.getRef());
  23.   device2.getRef().tell(PoisonPill.getInstance(), ActorRef.noSender());
  25.   DeviceGroup.RespondAllTemperatures response = requester.expectMsgClass(DeviceGroup.RespondAllTemperatures.class);
  26.   assertEquals(1L, response.requestId);
  28.   Map<String, DeviceGroup.TemperatureReading> expectedTemperatures = new HashMap<>();
  29.   expectedTemperatures.put("device1", new DeviceGroup.Temperature(1.0));
  30.   expectedTemperatures.put("device2", new DeviceGroup.Temperature(2.0));
  32.   assertEquals(expectedTemperatures, response.temperatures);
  33. }

最后一种情况是，并非所有设备都能及时响应。为了保持我们的测试相对较快，我们将用较小的超时构造DeviceGroupQuery Actor：

  1. @Test
  2. public void testReturnDeviceTimedOutIfDeviceDoesNotAnswerInTime() {
  3.   TestKit requester = new TestKit(system);
  5.   TestKit device1 = new TestKit(system);
  6.   TestKit device2 = new TestKit(system);
  8.   Map<ActorRef, String> actorToDeviceId = new HashMap<>();
  9.   actorToDeviceId.put(device1.getRef(), "device1");
  10.   actorToDeviceId.put(device2.getRef(), "device2");
  12.   ActorRef queryActor = system.actorOf(DeviceGroupQuery.props(
  13.           actorToDeviceId,
  14.           1L,
  15.           requester.getRef(),
  16.           new FiniteDuration(1, TimeUnit.SECONDS)));
  18.   assertEquals(0L, device1.expectMsgClass(Device.ReadTemperature.class).requestId);
  19.   assertEquals(0L, device2.expectMsgClass(Device.ReadTemperature.class).requestId);
  21.   queryActor.tell(new Device.RespondTemperature(0L, Optional.of(1.0)), device1.getRef());
  23.   DeviceGroup.RespondAllTemperatures response = requester.expectMsgClass(
  24.           java.time.Duration.ofSeconds(5),
  25.           DeviceGroup.RespondAllTemperatures.class);
  26.   assertEquals(1L, response.requestId);
  28.   Map<String, DeviceGroup.TemperatureReading> expectedTemperatures = new HashMap<>();
  29.   expectedTemperatures.put("device1", new DeviceGroup.Temperature(1.0));
  30.   expectedTemperatures.put("device2", DeviceGroup.DeviceTimedOut.INSTANCE);
  32.   assertEquals(expectedTemperatures, response.temperatures);
  33. }

查询功能已经按预期工作了，现在是时候在DeviceGroup Actor 中添加这个新功能了。

向设备组添加查询功能

现在在设备组 Actor 中包含查询功能相当简单。我们在查询 Actor 本身中完成了所有繁重的工作，设备组 Actor 只需要使用正确的初始参数创建它，而不需要其他任何参数。

  1. public class DeviceGroup extends AbstractActor {
  2.   private final LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);
  4.   final String groupId;
  6.   public DeviceGroup(String groupId) {
  7.     this.groupId = groupId;
  8.   }
  10.   public static Props props(String groupId) {
  11.     return Props.create(DeviceGroup.class, () -> new DeviceGroup(groupId));
  12.   }
  14.   public static final class RequestDeviceList {
  15.     final long requestId;
  17.     public RequestDeviceList(long requestId) {
  18.       this.requestId = requestId;
  19.     }
  20.   }
  22.   public static final class ReplyDeviceList {
  23.     final long requestId;
  24.     final Set<String> ids;
  26.     public ReplyDeviceList(long requestId, Set<String> ids) {
  27.       this.requestId = requestId;
  28.       this.ids = ids;
  29.     }
  30.   }
  32.   public static final class RequestAllTemperatures {
  33.     final long requestId;
  35.     public RequestAllTemperatures(long requestId) {
  36.       this.requestId = requestId;
  37.     }
  38.   }
  40.   public static final class RespondAllTemperatures {
  41.     final long requestId;
  42.     final Map<String, TemperatureReading> temperatures;
  44.     public RespondAllTemperatures(long requestId, Map<String, TemperatureReading> temperatures) {
  45.       this.requestId = requestId;
  46.       this.temperatures = temperatures;
  47.     }
  48.   }
  50.   public static interface TemperatureReading {
  51.   }
  53.   public static final class Temperature implements TemperatureReading {
  54.     public final double value;
  56.     public Temperature(double value) {
  57.       this.value = value;
  58.     }
  60.     @Override
  61.     public boolean equals(Object o) {
  62.       if (this == o) return true;
  63.       if (o == null || getClass() != o.getClass()) return false;
  65.       Temperature that = (Temperature) o;
  67.       return Double.compare(that.value, value) == 0;
  68.     }
  70.     @Override
  71.     public int hashCode() {
  72.       long temp = Double.doubleToLongBits(value);
  73.       return (int) (temp ^ (temp >>> 32));
  74.     }
  76.     @Override
  77.     public String toString() {
  78.       return "Temperature{" +
  79.         "value=" + value +
  80.         '}';
  81.     }
  82.   }
  84.   public enum TemperatureNotAvailable implements TemperatureReading {
  85.     INSTANCE
  86.   }
  88.   public enum DeviceNotAvailable implements TemperatureReading {
  89.     INSTANCE
  90.   }
  92.   public enum DeviceTimedOut implements TemperatureReading {
  93.     INSTANCE
  94.   }
  97.   final Map<String, ActorRef> deviceIdToActor = new HashMap<>();
  98.   final Map<ActorRef, String> actorToDeviceId = new HashMap<>();
  100.   @Override
  101.   public void preStart() {
  102.     log.info("DeviceGroup {} started", groupId);
  103.   }
  105.   @Override
  106.   public void postStop() {
  107.     log.info("DeviceGroup {} stopped", groupId);
  108.   }
  111.   private void onAllTemperatures(RequestAllTemperatures r) {
  112.     // since Java collections are mutable, we want to avoid sharing them between actors (since multiple Actors (threads)
  113.     // modifying the same mutable data-structure is not safe), and perform a defensive copy of the mutable map:
  114.     //
  115.     // Feel free to use your favourite immutable data-structures library with Akka in Java applications!
  116.     Map<ActorRef, String> actorToDeviceIdCopy = new HashMap<>(this.actorToDeviceId);
  118.     getContext().actorOf(DeviceGroupQuery.props(
  119.         actorToDeviceIdCopy, r.requestId, getSender(), new FiniteDuration(3, TimeUnit.SECONDS)));
  120.   }
  122.   @Override
  123.   public Receive createReceive() {
  124.             // ... other cases omitted
  125.             .match(RequestAllTemperatures.class, this::onAllTemperatures)
  126.             .build();
  127.   }
  128. }

或许值得重述一下我们在本章开头所说的话。通过将只与查询本身相关的临时状态保留在单独的 Actor 中，我们使组 Actor 的实现非常简单。它将一切委托给子 Actor，因此不必保留与核心业务无关的状态。此外，多个查询现在可以彼此并行运行，事实上，可以根据需要运行任意多个查询。在我们的例子中，查询单个设备 Actor 是一种快速操作，但是如果不是这样，例如，因为需要通过网络联系远程传感器，这种设计将显著提高吞吐量。

我们通过测试所有的功能一起工作来结束这一章。此测试是前一个测试的变体，现在使用组查询功能：

  1. @Test
  2. public void testCollectTemperaturesFromAllActiveDevices() {
  3.   TestKit probe = new TestKit(system);
  4.   ActorRef groupActor = system.actorOf(DeviceGroup.props("group"));
  6.   groupActor.tell(new DeviceManager.RequestTrackDevice("group", "device1"), probe.getRef());
  7.   probe.expectMsgClass(DeviceManager.DeviceRegistered.class);
  8.   ActorRef deviceActor1 = probe.getLastSender();
  10.   groupActor.tell(new DeviceManager.RequestTrackDevice("group", "device2"), probe.getRef());
  11.   probe.expectMsgClass(DeviceManager.DeviceRegistered.class);
  12.   ActorRef deviceActor2 = probe.getLastSender();
  14.   groupActor.tell(new DeviceManager.RequestTrackDevice("group", "device3"), probe.getRef());
  15.   probe.expectMsgClass(DeviceManager.DeviceRegistered.class);
  16.   ActorRef deviceActor3 = probe.getLastSender();
  18.   // Check that the device actors are working
  19.   deviceActor1.tell(new Device.RecordTemperature(0L, 1.0), probe.getRef());
  20.   assertEquals(0L, probe.expectMsgClass(Device.TemperatureRecorded.class).requestId);
  21.   deviceActor2.tell(new Device.RecordTemperature(1L, 2.0), probe.getRef());
  22.   assertEquals(1L, probe.expectMsgClass(Device.TemperatureRecorded.class).requestId);
  23.   // No temperature for device 3
  25.   groupActor.tell(new DeviceGroup.RequestAllTemperatures(0L), probe.getRef());
  26.   DeviceGroup.RespondAllTemperatures response = probe.expectMsgClass(DeviceGroup.RespondAllTemperatures.class);
  27.   assertEquals(0L, response.requestId);
  29.   Map<String, DeviceGroup.TemperatureReading> expectedTemperatures = new HashMap<>();
  30.   expectedTemperatures.put("device1", new DeviceGroup.Temperature(1.0));
  31.   expectedTemperatures.put("device2", new DeviceGroup.Temperature(2.0));
  32.   expectedTemperatures.put("device3", DeviceGroup.TemperatureNotAvailable.INSTANCE);
  34.   assertEquals(expectedTemperatures, response.temperatures);
  35. }

总结

在物联网（IoT）系统的背景下，本指南介绍了以下概念。如有必要，你可以通过以下链接进行查看：

下一步是什么？

要继续你的 Akka 之旅，我们建议：

  • 开始用 Akka 构建你自己的应用程序，如果你陷入困境的话，希望你能参与到我们的「社区」中，寻求帮助。
  • 如果你想了解更多的背景知识，请阅读参考文件的其余部分，并查看一些关于 Akka 的「书籍和视频」。

要从本指南获得完整的应用程序，你可能需要提供 UI 或 API。为此，我们建议你查看以下技术，看看哪些适合你：

  • 「Akka HTTP」是一个 HTTP 服务和客户端的库，使发布和使用 HTTP 端点（endpoints）成为可能。
  • 「Play Framework」是一个成熟的 Web 框架，它构建在 Akka HTTP 之上，它能与 Akka 很好地集成，可用于创建一个完整的现代 Web 用户界面。
  • 「Lagom」是一个基于 Akka 的独立的微服务框架，它编码了 Akka 和 Play 的许多最佳实践。

英文原文链接：Part 5: Querying Device Groups.