第 4 部分: 使用设备组

依赖

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

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"

简介

让我们仔细看看用例所需的主要功能。在用于监测家庭温度的完整物联网系统中，将设备传感器连接到系统的步骤可能如下：

1. 家庭中的传感器设备通过某种协议进行连接。
2. 管理网络连接的组件接受连接。
3. 传感器提供其组和设备 ID，以便在系统的设备管理器组件中注册。
4. 设备管理器组件通过查找或创建负责保持传感器状态的 Actor 来处理注册。
5. Actor 以一种确认（acknowledgement）回应，暴露其ActorRef。
6. 网络组件现在使用ActorRef在传感器和设备 Actor 之间进行通信，而不需要经过设备管理器。

步骤 1 和 2 发生在教程系统的边界之外。在本章中，我们将开始处理步骤 3 - 6，并创建传感器在系统中注册和与 Actor 通信的方法。但首先，我们有另一个体系结构决策——我们应该使用多少个层次的 Actor 来表示设备组和设备传感器？

Akka 程序员面临的主要设计挑战之一是为 Actor 选择最佳的粒度。在实践中，根据 Actor 之间交互的特点，通常有几种有效的方法来组织系统。例如，在我们的用例中，可能有一个 Actor 维护所有的组和设备——或许可以使用哈希表（hash maps）。对于每个跟踪同一个家中所有设备状态的组来说，有一个 Actor 也是合理的。

以下指导原则可以帮助我们选择最合适的 Actor 层次结构：

• 一般来说，更倾向于更大的粒度。引入比需要更多的细粒度 Actor 会导致比它解决的问题更多的问题。
• 当系统需要时添加更细的粒度：
• 更高的并发性。
• 有许多状态的 Actor 之间的复杂交互。在下一章中，我们将看到一个很好的例子。
• 足够多的状态，划分为较小的 Actor 是有意义地。
• 多重无关责任。使用不同的 Actor 可以使单个 Actor 失败并恢复，而对其他的 Actor 影响很小。

设备管理器层次结构

考虑到上一节中概述的原则，我们将设备管理器组件建模为具有三个级别的 Actor 树：

• 顶级监督者 Actor 表示设备的系统组件。它也是查找和创建设备组和设备 Actor 的入口点。
• 在下一个级别，每个组 Actor 都监督设备 Actor 使用同一个组 ID。它们还提供服务，例如查询组中所有可用设备的温度读数。
• 设备 Actor 管理与实际设备传感器的所有交互，例如存储温度读数。

我们选择这三层架构的原因如下：

• 划分组为单独的 Actor：
• 隔离组中发生的故障。如果一个 Actor 管理所有设备组，则一个组中导致重新启动的错误将清除组的状态，否则这些组不会出现故障。
• 简化了查询属于一个组的所有设备的问题。每个组 Actor 只包含与其组相关的状态。
• 提高系统的并行性。因为每个组都有一个专用的 Actor，所以它们可以并发运行，我们可以并发查询多个组。
• 将传感器建模为单个设备 Actor：
• 将一个设备 Actor 的故障与组中的其他设备隔离开来。
• 增加收集温度读数的平行度。来自不同传感器的网络连接直接与各自的设备 Actor 通信，从而减少了竞争点。

定义了设备体系结构后，我们就可以开始研究注册传感器的协议了。

注册协议

作为第一步，我们需要设计协议来注册一个设备，以及创建负责它的组和设备 Actor。此协议将由DeviceManager组件本身提供，因为它是唯一已知且预先可用的 Actor：设备组和设备 Actor 是按需创建的。

更详细地看一下注册，我们可以概述必要的功能：

• 当DeviceManager接收到具有组和设备 ID 的请求时：
• 如果管理器已经有了设备组的 Actor，那么它会将请求转发给它。
• 否则，它会创建一个新的设备组 Actor，然后转发请求。
• DeviceGroup Actor 接收为给定设备注册 Actor 的请求：
• 如果组已经有设备的 Actor，则组 Actor 将请求转发给设备 Actor。
• 否则，设备组 Actor 首先创建设备 Actor，然后转发请求。
• 设备 Actor 接收请求并向原始发送者发送确认。由于设备 Actor 确认接收（而不是组 Actor），传感器现在将有ActorRef，可以直接向其 Actor 发送消息。

我们将用来传递注册请求及其确认的消息有一个简单的定义：

1. public static final class RequestTrackDevice {
2. public final String groupId;
3. public final String deviceId;

5. public RequestTrackDevice(String groupId, String deviceId) {
6. this.groupId = groupId;
7. this.deviceId = deviceId;
8. }
9. }

11. public static final class DeviceRegistered {
12. }

在这种情况下，我们在消息中没有包含请求 ID 字段。由于注册只发生一次，当组件将系统连接到某个网络协议时，ID 并不重要。但是，包含请求 ID 通常是一种最佳实践。

现在，我们将从头开始实现该协议。在实践中，自上向下和自下而上的方法都很有效，但是在我们的例子中，我们实使用自下而上的方法，因为它允许我们立即为新特性编写测试，而不需要模拟出稍后需要构建的部分。

向设备 Actor 添加注册支持

在我们的层次结构的底部是Device Actor。他们在注册过程中的工作很简单：回复注册请求并向发送者确认。对于带有不匹配的组或设备 ID 的请求，添加一个保护措施也是明智的。

我们假设注册消息发送者的 ID 保留在上层。我们将在下一节向你展示如何实现这一点。

设备 Actor 的注册代码如下所示：

1. import akka.actor.AbstractActor;
2. import akka.actor.Props;
3. import akka.event.Logging;
4. import akka.event.LoggingAdapter;

6. import jdocs.tutorial_4.DeviceManager.DeviceRegistered;
7. import jdocs.tutorial_4.DeviceManager.RequestTrackDevice;

9. import java.util.Optional;

11. public class Device extends AbstractActor {
12. private final LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);

14. final String groupId;

16. final String deviceId;

18. public Device(String groupId, String deviceId) {
19. this.groupId = groupId;
20. this.deviceId = deviceId;
21. }

23. public static Props props(String groupId, String deviceId) {
24. return Props.create(Device.class, () -> new Device(groupId, deviceId));
25. }

27. public static final class RecordTemperature {
28. final long requestId;
29. final double value;

31. public RecordTemperature(long requestId, double value) {
32. this.requestId = requestId;
33. this.value = value;
34. }
35. }

37. public static final class TemperatureRecorded {
38. final long requestId;

40. public TemperatureRecorded(long requestId) {
41. this.requestId = requestId;
42. }
43. }

45. public static final class ReadTemperature {
46. final long requestId;

48. public ReadTemperature(long requestId) {
49. this.requestId = requestId;
50. }
51. }

53. public static final class RespondTemperature {
54. final long requestId;
55. final Optional<Double> value;

57. public RespondTemperature(long requestId, Optional<Double> value) {
58. this.requestId = requestId;
59. this.value = value;
60. }
61. }

63. Optional<Double> lastTemperatureReading = Optional.empty();

65. @Override
66. public void preStart() {
67. log.info("Device actor {}-{} started", groupId, deviceId);
68. }

70. @Override
71. public void postStop() {
72. log.info("Device actor {}-{} stopped", groupId, deviceId);
73. }

75. @Override
76. public Receive createReceive() {
77. return receiveBuilder()
78. .match(RequestTrackDevice.class, r -> {
79. if (this.groupId.equals(r.groupId) && this.deviceId.equals(r.deviceId)) {
80. getSender().tell(new DeviceRegistered(), getSelf());
81. } else {
82. log.warning(
83. "Ignoring TrackDevice request for {}-{}.This actor is responsible for {}-{}.",
84. r.groupId, r.deviceId, this.groupId, this.deviceId
85. );
86. }
87. })
88. .match(RecordTemperature.class, r -> {
89. log.info("Recorded temperature reading {} with {}", r.value, r.requestId);
90. lastTemperatureReading = Optional.of(r.value);
91. getSender().tell(new TemperatureRecorded(r.requestId), getSelf());
92. })
93. .match(ReadTemperature.class, r -> {
94. getSender().tell(new RespondTemperature(r.requestId, lastTemperatureReading), getSelf());
95. })
96. .build();
97. }
98. }

我们现在可以编写两个新的测试用例，一个成功注册，另一个在 ID 不匹配时测试用例：

1. @Test
2. public void testReplyToRegistrationRequests() {
3. TestKit probe = new TestKit(system);
4. ActorRef deviceActor = system.actorOf(Device.props("group", "device"));

6. deviceActor.tell(new DeviceManager.RequestTrackDevice("group", "device"), probe.getRef());
7. probe.expectMsgClass(DeviceManager.DeviceRegistered.class);
8. assertEquals(deviceActor, probe.getLastSender());
9. }

11. @Test
12. public void testIgnoreWrongRegistrationRequests() {
13. TestKit probe = new TestKit(system);
14. ActorRef deviceActor = system.actorOf(Device.props("group", "device"));

16. deviceActor.tell(new DeviceManager.RequestTrackDevice("wrongGroup", "device"), probe.getRef());
17. probe.expectNoMessage();

19. deviceActor.tell(new DeviceManager.RequestTrackDevice("group", "wrongDevice"), probe.getRef());
20. probe.expectNoMessage();
21. }

• 注释：我们使用了TestKit中的expectNoMsg()帮助者方法。此断言等待到定义的时间限制，如果在此期间收到任何消息，则会失败。如果在等待期间未收到任何消息，则断言通过。通常最好将这些超时保持在较低的水平（但不要太低），因为它们会增加大量的测试执行时间。

向设备组 Actor 添加注册支持

我们已经完成了设备级别的注册支持，现在我们必须在组级别实现它。当涉及到注册时，组 Actor 有更多的工作要做，包括：

• 通过将注册请求转发给现有设备 Actor 或创建新 Actor 并转发消息来处理注册请求。
• 跟踪组中存在哪些设备 Actor，并在停止时将其从组中删除。

处理注册请求

设备组 Actor 必须将请求转发给现有的子 Actor，或者应该创建一个子 Actor。要通过设备 ID 查找子 Actor，我们将使用Map<String, ActorRef>。

我们还希望保留请求的原始发送者的 ID，以便设备 Actor 可以直接回复。这可以通过使用forward而不是tell运算符来实现。两者之间的唯一区别是，forward保留原始发送者，而tell将发送者设置为当前 Actor。就像我们的设备 Actor 一样，我们确保不响应错误的组 ID。将以下内容添加到你的源文件中：

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. final Map<String, ActorRef> deviceIdToActor = new HashMap<>();

16. @Override
17. public void preStart() {
18. log.info("DeviceGroup {} started", groupId);
19. }

21. @Override
22. public void postStop() {
23. log.info("DeviceGroup {} stopped", groupId);
24. }

26. private void onTrackDevice(DeviceManager.RequestTrackDevice trackMsg) {
27. if (this.groupId.equals(trackMsg.groupId)) {
28. ActorRef deviceActor = deviceIdToActor.get(trackMsg.deviceId);
29. if (deviceActor != null) {
30. deviceActor.forward(trackMsg, getContext());
31. } else {
32. log.info("Creating device actor for {}", trackMsg.deviceId);
33. deviceActor = getContext().actorOf(Device.props(groupId, trackMsg.deviceId), "device-" + trackMsg.deviceId);
34. deviceIdToActor.put(trackMsg.deviceId, deviceActor);
35. deviceActor.forward(trackMsg, getContext());
36. }
37. } else {
38. log.warning(
39. "Ignoring TrackDevice request for {}. This actor is responsible for {}.",
40. groupId, this.groupId
41. );
42. }
43. }

45. @Override
46. public Receive createReceive() {
47. return receiveBuilder()
48. .match(DeviceManager.RequestTrackDevice.class, this::onTrackDevice)
49. .build();
50. }
51. }

正如我们对设备所做的那样，我们测试了这个新功能。我们还测试了两个不同 ID 返回的 Actor 实际上是不同的，我们还尝试记录每个设备的温度读数，以查看 Actor 是否有响应。

1. @Test
2. public void testRegisterDeviceActor() {
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();
13. assertNotEquals(deviceActor1, deviceActor2);

15. // Check that the device actors are working
16. deviceActor1.tell(new Device.RecordTemperature(0L, 1.0), probe.getRef());
17. assertEquals(0L, probe.expectMsgClass(Device.TemperatureRecorded.class).requestId);
18. deviceActor2.tell(new Device.RecordTemperature(1L, 2.0), probe.getRef());
19. assertEquals(1L, probe.expectMsgClass(Device.TemperatureRecorded.class).requestId);
20. }

22. @Test
23. public void testIgnoreRequestsForWrongGroupId() {
24. TestKit probe = new TestKit(system);
25. ActorRef groupActor = system.actorOf(DeviceGroup.props("group"));

27. groupActor.tell(new DeviceManager.RequestTrackDevice("wrongGroup", "device1"), probe.getRef());
28. probe.expectNoMessage();
29. }

如果注册请求已经存在设备 Actor，我们希望使用现有的 Actor 而不是新的 Actor。我们尚未对此进行测试，因此需要修复此问题：

1. @Test
2. public void testReturnSameActorForSameDeviceId() {
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", "device1"), probe.getRef());
11. probe.expectMsgClass(DeviceManager.DeviceRegistered.class);
12. ActorRef deviceActor2 = probe.getLastSender();
13. assertEquals(deviceActor1, deviceActor2);
14. }

跟踪组内的设备 Actor

到目前为止，我们已经实现了在组中注册设备 Actor 的逻辑。然而，设备增增减减（come and go），所以我们需要一种方法从Map<String, ActorRef>中删除设备 Actor。我们假设当一个设备被删除时，它对应的设备 Actor 被停止。正如我们前面讨论的，监督只处理错误场景——而不是优雅的停止。因此，当其中一个设备 Actor 停止时，我们需要通知其父 Actor。

Akka 提供了一个死亡观察功能（Death Watch feature），允许一个 Actor 观察另一个 Actor，并在另一个 Actor 被停止时得到通知。与监督者不同的是，观察（watching）并不局限于父子关系，任何 Actor 只要知道ActorRef就可以观察其他 Actor。在被观察的 Actor 停止后，观察者接收一条Terminated(actorRef)消息，该消息还包含对被观察的 Actor 的引用。观察者可以显式处理此消息，也可以失败并出现DeathPactException。如果在被观察的 Actor 被停止后，该 Actor 不能再履行自己的职责，则后者很有用。在我们的例子中，组应该在一个设备停止后继续工作，所以我们需要处理Terminated(actorRef)消息。

我们的设备组 Actor 需要包括以下功能：

• 当新设备 Actor 被创建时开始观察（watching）。
• 当通知指示设备已停止时，从映射Map<String, ActorRef>中删除设备 Actor。

不幸的是，Terminated的消息只包含子 Actor 的ActorRef。我们需要 Actor 的 ID 将其从现有设备到设备的 Actor 映射中删除。为了能够进行删除，我们需要引入另一个占位符Map<ActorRef, String>，它允许我们找到与给定ActorRef对应的设备 ID。

添加用于标识 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. final Map<String, ActorRef> deviceIdToActor = new HashMap<>();
15. final Map<ActorRef, String> actorToDeviceId = new HashMap<>();

17. @Override
18. public void preStart() {
19. log.info("DeviceGroup {} started", groupId);
20. }

22. @Override
23. public void postStop() {
24. log.info("DeviceGroup {} stopped", groupId);
25. }

27. private void onTrackDevice(DeviceManager.RequestTrackDevice trackMsg) {
28. if (this.groupId.equals(trackMsg.groupId)) {
29. ActorRef deviceActor = deviceIdToActor.get(trackMsg.deviceId);
30. if (deviceActor != null) {
31. deviceActor.forward(trackMsg, getContext());
32. } else {
33. log.info("Creating device actor for {}", trackMsg.deviceId);
34. deviceActor = getContext().actorOf(Device.props(groupId, trackMsg.deviceId), "device-" + trackMsg.deviceId);
35. getContext().watch(deviceActor);
36. actorToDeviceId.put(deviceActor, trackMsg.deviceId);
37. deviceIdToActor.put(trackMsg.deviceId, deviceActor);
38. deviceActor.forward(trackMsg, getContext());
39. }
40. } else {
41. log.warning(
42. "Ignoring TrackDevice request for {}. This actor is responsible for {}.",
43. groupId, this.groupId
44. );
45. }
46. }

48. private void onTerminated(Terminated t) {
49. ActorRef deviceActor = t.getActor();
50. String deviceId = actorToDeviceId.get(deviceActor);
51. log.info("Device actor for {} has been terminated", deviceId);
52. actorToDeviceId.remove(deviceActor);
53. deviceIdToActor.remove(deviceId);
54. }

56. @Override
57. public Receive createReceive() {
58. return receiveBuilder()
59. .match(DeviceManager.RequestTrackDevice.class, this::onTrackDevice)
60. .match(Terminated.class, this::onTerminated)
61. .build();
62. }
63. }

到目前为止，我们还没有办法获得组设备 Actor 跟踪的设备，因此，我们还不能测试我们的新功能。为了使其可测试，我们添加了一个新的查询功能（消息RequestDeviceList），其中列出了当前活动的设备 ID：

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. final Map<String, ActorRef> deviceIdToActor = new HashMap<>();
33. final Map<ActorRef, String> actorToDeviceId = new HashMap<>();

35. @Override
36. public void preStart() {
37. log.info("DeviceGroup {} started", groupId);
38. }

40. @Override
41. public void postStop() {
42. log.info("DeviceGroup {} stopped", groupId);
43. }

45. private void onTrackDevice(DeviceManager.RequestTrackDevice trackMsg) {
46. if (this.groupId.equals(trackMsg.groupId)) {
47. ActorRef deviceActor = deviceIdToActor.get(trackMsg.deviceId);
48. if (deviceActor != null) {
49. deviceActor.forward(trackMsg, getContext());
50. } else {
51. log.info("Creating device actor for {}", trackMsg.deviceId);
52. deviceActor = getContext().actorOf(Device.props(groupId, trackMsg.deviceId), "device-" + trackMsg.deviceId);
53. getContext().watch(deviceActor);
54. actorToDeviceId.put(deviceActor, trackMsg.deviceId);
55. deviceIdToActor.put(trackMsg.deviceId, deviceActor);
56. deviceActor.forward(trackMsg, getContext());
57. }
58. } else {
59. log.warning(
60. "Ignoring TrackDevice request for {}. This actor is responsible for {}.",
61. groupId, this.groupId
62. );
63. }
64. }

66. private void onDeviceList(RequestDeviceList r) {
67. getSender().tell(new ReplyDeviceList(r.requestId, deviceIdToActor.keySet()), getSelf());
68. }

70. private void onTerminated(Terminated t) {
71. ActorRef deviceActor = t.getActor();
72. String deviceId = actorToDeviceId.get(deviceActor);
73. log.info("Device actor for {} has been terminated", deviceId);
74. actorToDeviceId.remove(deviceActor);
75. deviceIdToActor.remove(deviceId);
76. }

78. @Override
79. public Receive createReceive() {
80. return receiveBuilder()
81. .match(DeviceManager.RequestTrackDevice.class, this::onTrackDevice)
82. .match(RequestDeviceList.class, this::onDeviceList)
83. .match(Terminated.class, this::onTerminated)
84. .build();
85. }
86. }

我们几乎准备好测试设备的移除功能了。但是，我们仍然需要以下功能：

• 为了通过我们的测试用例停止一个设备 Actor。从外面看，任何 Actor 都可以通过发送一个特殊的内置消息PoisonPill来停止，该消息指示 Actor 停止。
• 为了在设备 Actor 停止后得到通知。我们也可以使用Death Watch功能观察设备。TestKit有两条消息，我们可以很容易地使用watch()来观察指定的 Actor，使用expectTerminated来断言被观察的 Actor 已被终止。

我们现在再添加两个测试用例。在第一个测试中，我们测试在添加了一些设备之后，是否能返回正确的 ID 列表。第二个测试用例确保在设备 Actor 停止后正确删除设备 ID：

1. @Test
2. public void testListActiveDevices() {
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);

9. groupActor.tell(new DeviceManager.RequestTrackDevice("group", "device2"), probe.getRef());
10. probe.expectMsgClass(DeviceManager.DeviceRegistered.class);

12. groupActor.tell(new DeviceGroup.RequestDeviceList(0L), probe.getRef());
13. DeviceGroup.ReplyDeviceList reply = probe.expectMsgClass(DeviceGroup.ReplyDeviceList.class);
14. assertEquals(0L, reply.requestId);
15. assertEquals(Stream.of("device1", "device2").collect(Collectors.toSet()), reply.ids);
16. }

18. @Test
19. public void testListActiveDevicesAfterOneShutsDown() {
20. TestKit probe = new TestKit(system);
21. ActorRef groupActor = system.actorOf(DeviceGroup.props("group"));

23. groupActor.tell(new DeviceManager.RequestTrackDevice("group", "device1"), probe.getRef());
24. probe.expectMsgClass(DeviceManager.DeviceRegistered.class);
25. ActorRef toShutDown = probe.getLastSender();

27. groupActor.tell(new DeviceManager.RequestTrackDevice("group", "device2"), probe.getRef());
28. probe.expectMsgClass(DeviceManager.DeviceRegistered.class);

30. groupActor.tell(new DeviceGroup.RequestDeviceList(0L), probe.getRef());
31. DeviceGroup.ReplyDeviceList reply = probe.expectMsgClass(DeviceGroup.ReplyDeviceList.class);
32. assertEquals(0L, reply.requestId);
33. assertEquals(Stream.of("device1", "device2").collect(Collectors.toSet()), reply.ids);

35. probe.watch(toShutDown);
36. toShutDown.tell(PoisonPill.getInstance(), ActorRef.noSender());
37. probe.expectTerminated(toShutDown);

39. // using awaitAssert to retry because it might take longer for the groupActor
40. // to see the Terminated, that order is undefined
41. probe.awaitAssert(() -> {
42. groupActor.tell(new DeviceGroup.RequestDeviceList(1L), probe.getRef());
43. DeviceGroup.ReplyDeviceList r =
44. probe.expectMsgClass(DeviceGroup.ReplyDeviceList.class);
45. assertEquals(1L, r.requestId);
46. assertEquals(Stream.of("device2").collect(Collectors.toSet()), r.ids);
47. return null;
48. });
49. }

创建设备管理器 Actor

在我们的层次结构中，我们需要在DeviceManager源文件中为设备管理器组件创建入口点。此 Actor 与设备组 Actor 非常相似，但创建的是设备组 Actor 而不是设备 Actor：

1. public class DeviceManager extends AbstractActor {
2. private final LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);

4. public static Props props() {
5. return Props.create(DeviceManager.class, DeviceManager::new);
6. }

8. public static final class RequestTrackDevice {
9. public final String groupId;
10. public final String deviceId;

12. public RequestTrackDevice(String groupId, String deviceId) {
13. this.groupId = groupId;
14. this.deviceId = deviceId;
15. }
16. }

18. public static final class DeviceRegistered {
19. }

21. final Map<String, ActorRef> groupIdToActor = new HashMap<>();
22. final Map<ActorRef, String> actorToGroupId = new HashMap<>();

24. @Override
25. public void preStart() {
26. log.info("DeviceManager started");
27. }

29. @Override
30. public void postStop() {
31. log.info("DeviceManager stopped");
32. }

34. private void onTrackDevice(RequestTrackDevice trackMsg) {
35. String groupId = trackMsg.groupId;
36. ActorRef ref = groupIdToActor.get(groupId);
37. if (ref != null) {
38. ref.forward(trackMsg, getContext());
39. } else {
40. log.info("Creating device group actor for {}", groupId);
41. ActorRef groupActor = getContext().actorOf(DeviceGroup.props(groupId), "group-" + groupId);
42. getContext().watch(groupActor);
43. groupActor.forward(trackMsg, getContext());
44. groupIdToActor.put(groupId, groupActor);
45. actorToGroupId.put(groupActor, groupId);
46. }
47. }

49. private void onTerminated(Terminated t) {
50. ActorRef groupActor = t.getActor();
51. String groupId = actorToGroupId.get(groupActor);
52. log.info("Device group actor for {} has been terminated", groupId);
53. actorToGroupId.remove(groupActor);
54. groupIdToActor.remove(groupId);
55. }

57. public Receive createReceive() {
58. return receiveBuilder()
59. .match(RequestTrackDevice.class, this::onTrackDevice)
60. .match(Terminated.class, this::onTerminated)
61. .build();
62. }
63. }

我们将设备管理器的测试留给你作为练习，因为它与我们为设备组 Actor 编写的测试非常相似。

下一步是什么？

我们现在有了一个用于注册和跟踪设备以及记录测量值的分层组件。我们已经了解了如何实现不同类型的对话模式，例如：

• 请求响应（Request-respond），用于温度记录。
• 代理响应（Delegate-respond），用于设备注册。
• 创建监视终止（Create-watch-terminate），用于将组和设备 Actor 创建为子级。

在下一章中，我们将介绍组查询功能，这将建立一种新的分散收集（scatter-gather）对话模式。特别地，我们将实现允许用户查询属于一个组的所有设备的状态的功能。

英文原文链接：Part 4: Working with Device Groups.