【AKKA 官方文档翻译】第四部分：使用设备组

第四部分：使用设备组

akka版本2.5.8

版权声明：本文为博主原创文章，未经博主允许不得转载。

让我们仔细观察下我们用例所要求的主要功能。在完整的检测家庭温度的物联网系统中，传感器设备和我们的系统进行连接的步骤大概会像这样：

1、家里的一个传感器设备通过某个协议发起连接

2、组件管理器处理网络连接并接受连接

3、传感器提供组ID和设备ID来向我们的系统组件管理器注册

4、设备管理器组件处理注册信息，并通过询问或创建actor负责任务的方式保持传感器状态

5、actor返回确认信息，并暴露其

ActorRef

6、网络组件使用

ActorRef

来实现传感器和设备actor的交流，而不用通过设备管理器

第一步和第二步在我们教程之外，在本章中，我们会开始讨论步骤3-6，并且创建一个让传感器可以与我们的系统注册和通信的方式。然而首先我们要讨论一个架构问题，我们需要用几个层次来表示设备组和传感器设备？

对于Akka程序员，其中一个主要的挑战就是给actor选择一个合适的粒度。在实践中，依赖于actor之间的特性，我们有几种有效的方式来组织系统。在我们的用例中，可能会存在单一actor来维护所有的组和设备——可能使用哈希映射。如果我们为每个组创建一个actor来维护状态也是合理的。

以下指南帮助我们选择合适的actor层次：

1、在通常情况下，我们偏向较大的粒度。引入过细粒度（超过需求）的actor可能引入的问题会超过其解决的问题。

2、当系统需要时添加更精细的粒度。

1、更高的并发性。

2、actor之间很多复杂的交流会拥有很多状态信息，我们会在下章里看到很多好的实例来解决它。

3、分成小的actor提供了足够的状态信息。

4、很多互相无关的责任。使用独立的actor可以在个体失败并恢复时对系统产生较小的影响。

设备管理器层级

考虑到上一节所述的原则，我们会将设备管理器建模为一个三层的树：

1、最高级监管actor代表设备的系统组件，它同样负责维持和创建设备组和设备actor

2、在下一级，各个组actor监管着属于他们的设备actor。他们也提供服务，例如向组里所有的设备请求温度数据

3、设备actor管理所有和实际传感器设备的交互，例如存储温度读数



我们使用三层架构主要原因是：

1、使单独的actor组成一个组

1、将组内的故障进行隔离，如果只有一个actor去管理所有的设备组，一旦一个组出现错误就会导致重启，并可能摧毁所有组的状态，不管其有没有出错

2、简化请求一个组内所有设备数据的问题，每个组actor只包含与组相关的状态

3、增加系统的并行性。每个组都有一个专用的actor，他们可以并行地运行，我们也可以并行地去取所有组的数据

2、将传感器建模为单个设备actor

1、隔离组内各个设备actor的错误

2、增加读取温度数据的并行度，每个传感器通过网络和它们对应的设备actor通信，减少网络争用点

在我们定义的架构下，我们可以开始处理传感器注册协议了。

注册协议

第一步，我们需要创建它所负责的协议：注册设备、创建组和创建设备actor。这个协议会被

DeviceManager

组件提供，因为它是唯一可以预先了解设备情况的———设备组和设备actor是按需创建的。

让我们来详细看下注册，我们可以大概列出必要功能：

1、当

DeviceManager

接收到一个包含组ID和设备ID的请求时：

1、如果管理者已经拥有了这个组的组actor，则将请求转发给它

2、否则创建一个组actor，然后转发请求给它

2、当

DeviceGroup

actor接收到设备的注册请求：

1、如果组内已经有了这个设备的actor，则将请求转发给这个actor

2、否则创建一个设备actor，然后转发请求给它

3、设备actor接收请求并发送回应给原始传感器。由于设备actor发送了确认信息（而不是组actor），传感器就可以持有

ActorRef

，并直接向它对应的设备actor发送消息

我们对用来请求注册和回应的消息做个简单的定义：

final case class RequestTrackDevice(groupId: String, deviceId: String)
case object DeviceRegistered

在这个场景下，我们没有在消息内包含ID字段，因为在注册时时ID并不重要。但是包含请求ID通常是最佳做法。

现在我们开始自下而上实现我们的协议。在实践中，自上而下和自下而上都是可以的，但是在我们的场景中，我们使用自下而上的设计将会很有用。这允许我们立即编写测试用例而不用模拟我们之后要编写的功能来测试新功能。

向设备actor添加注册支持

Device

actor在我们层级结构的底部，注册的工作是很简单的：回应注册请求给发送者。谨慎而言，我们需要对不匹配的组ID和设备ID做安全防护。

我们假定注册信息发送者的ID在上层被存储，我们将在下节来展示实现方式。

设备actor的注册代码看起来就像这样，修改你的示例代码来匹配它。

object Device {
def props(groupId: String, deviceId: String): Props = Props(new Device(groupId, deviceId))

final case class RecordTemperature(requestId: Long, value: Double)
final case class TemperatureRecorded(requestId: Long)

final case class ReadTemperature(requestId: Long)
final case class RespondTemperature(requestId: Long, value: Option[Double])
}

class Device(groupId: String, deviceId: String) extends Actor with ActorLogging {
import Device._

var lastTemperatureReading: Option[Double] = None

override def preStart(): Unit = log.info("Device actor {}-{} started", groupId, deviceId)

override def postStop(): Unit = log.info("Device actor {}-{} stopped", groupId, deviceId)

override def receive: Receive = {
case DeviceManager.RequestTrackDevice(`groupId`, `deviceId`) ⇒
sender() ! DeviceManager.DeviceRegistered

case DeviceManager.RequestTrackDevice(groupId, deviceId) ⇒
log.warning(
"Ignoring TrackDevice request for {}-{}.This actor is responsible for {}-{}.",
groupId, deviceId, this.groupId, this.deviceId
)

case RecordTemperature(id, value) ⇒
log.info("Recorded temperature reading {} with {}", value, id)
lastTemperatureReading = Some(value)
sender() ! TemperatureRecorded(id)

case ReadTemperature(id) ⇒
sender() ! RespondTemperature(id, lastTemperatureReading)
}
}

注意：

我们使用了scala模式匹配的特性来检查某个字段是不是我们的期望值。通过使用反引号把变量括起来

variable

,模式只有在值一致时才会匹配。

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

"reply to registration requests" in {
val probe = TestProbe()
val deviceActor = system.actorOf(Device.props("group", "device"))

deviceActor.tell(DeviceManager.RequestTrackDevice("group", "device"), probe.ref)
probe.expectMsg(DeviceManager.DeviceRegistered)
probe.lastSender should ===(deviceActor)
}

"ignore wrong registration requests" in {
val probe = TestProbe()
val deviceActor = system.actorOf(Device.props("group", "device"))

deviceActor.tell(DeviceManager.RequestTrackDevice("wrongGroup", "device"), probe.ref)
probe.expectNoMsg(500.milliseconds)

deviceActor.tell(DeviceManager.RequestTrackDevice("group", "Wrongdevice"), probe.ref)
probe.expectNoMsg(500.milliseconds)
}

注意：

在

TestProbe

中我们使用了帮助方法

expectNoMsg()

，这个方法等待指定的时间，如果在限制时间内接收到消息则失败，如果超时则断言成功。使用这些超时（但不能太低）是个好方法，但是它增加了测试的执行时间。

向设备组actor添加注册支持

我们已经完成了设备级别的注册支持，现在我们来实现组级别的注册支持。一个组actor在接收到注册请求时有很多工作要做，包括：

1、处理注册请求。转发给已经存在的设备actor或者创建新的actor并且将请求转发给它

2、跟踪组内的设备actor，当它们停止时从组内移除它们

处理注册请求

一个设备组actor必须转发请求或者创建一个actor。为了通过设备ID查找子actor，我们需要一个

Map[String, ActorRef]

。

我们还想保留原始请求发件人的ID，以便我们的设备actor可以直接回复。这可以通过使用

forward

而不是

!

来实现。两者唯一的不同之处就是

forward

保持了原始的发件人信息，而

!

把当前actor作为发件人。就像我们的设备actor一样，我们保证不会回复错误的组ID请求。添加以下文件到你的源文件中：

object DeviceGroup {
def props(groupId: String): Props = Props(new DeviceGroup(groupId))
}

class DeviceGroup(groupId: String) extends Actor with ActorLogging {
var deviceIdToActor = Map.empty[String, ActorRef]

override def preStart(): Unit = log.info("DeviceGroup {} started", groupId)

override def postStop(): Unit = log.info("DeviceGroup {} stopped", groupId)

override def receive: Receive = {
case trackMsg @ RequestTrackDevice(`groupId`, _) ⇒
deviceIdToActor.get(trackMsg.deviceId) match {
case Some(deviceActor) ⇒
deviceActor forward trackMsg
case None ⇒
log.info("Creating device actor for {}", trackMsg.deviceId)
val deviceActor = context.actorOf(Device.props(groupId, trackMsg.deviceId), s"device-${trackMsg.deviceId}")
deviceIdToActor += trackMsg.deviceId -> deviceActor
deviceActor forward trackMsg
}

case RequestTrackDevice(groupId, deviceId) ⇒
log.warning(
"Ignoring TrackDevice request for {}. This actor is responsible for {}.",
groupId, this.groupId
)
}
}

就像我们对设备所做的一样，我们测试新功能。我们还测试了actor对于不同的ID的回复是不同的，并且我们还尝试记录每个设备读取的温度，以查看actor是否响应。

"be able to register a device actor" in {
val probe = TestProbe()
val groupActor = system.actorOf(DeviceGroup.props("group"))

groupActor.tell(DeviceManager.RequestTrackDevice("group", "device1"), probe.ref)
probe.expectMsg(DeviceManager.DeviceRegistered)
val deviceActor1 = probe.lastSender

groupActor.tell(DeviceManager.RequestTrackDevice("group", "device2"), probe.ref)
probe.expectMsg(DeviceManager.DeviceRegistered)
val deviceActor2 = probe.lastSender
deviceActor1 should !==(deviceActor2)

// Check that the device actors are working
deviceActor1.tell(Device.RecordTemperature(requestId = 0, 1.0), probe.ref)
probe.expectMsg(Device.TemperatureRecorded(requestId = 0))
deviceActor2.tell(Device.RecordTemperature(requestId = 1, 2.0), probe.ref)
probe.expectMsg(Device.TemperatureRecorded(requestId = 1))
}

"ignore requests for wrong groupId" in {
val probe = TestProbe()
val groupActor = system.actorOf(DeviceGroup.props("group"))

groupActor.tell(DeviceManager.RequestTrackDevice("wrongGroup", "device1"), probe.ref)
probe.expectNoMsg(500.milliseconds)
}

在注册请求中，如果设备actor已经存在，我们会使用现有的这个actor而不是再新建一个actor。我们还没有测试这个功能，所以我们修改一下：

"return same actor for same deviceId" in {
val probe = TestProbe()
val groupActor = system.actorOf(DeviceGroup.props("group"))

groupActor.tell(DeviceManager.RequestTrackDevice("group", "device1"), probe.ref)
probe.expectMsg(DeviceManager.DeviceRegistered)
val deviceActor1 = probe.lastSender

groupActor.tell(DeviceManager.RequestTrackDevice("group", "device1"), probe.ref)
probe.expectMsg(DeviceManager.DeviceRegistered)
val deviceActor2 = probe.lastSender

deviceActor1 should ===(deviceActor2)
}

跟踪组内的设备actor

到目前为止，我们已经实现了向组注册设备actor的逻辑。设备可以放置和去除，所以我们需要一种方式从

Map[String, ActorRef]

删除设备actor。我们假设当设备被去除时，它对应的设备actor也会简单地停止。我们之前讲过监管者，但是它只能处理错
bf6b
误的情况，正常停止是没有办法的。所以我们需要在设备actor停止时通知父actor。

Akka提供了一个死亡观察功能，它允许一个actor去观察另一个actor，当另一个actor停止时，这个actor会被通知。和监管者不同，这种观察不局限于父子关系，在拥有其

ActorRef

的情况下，所有的actor都可以观察任何actor。当被观察的actor停止时，观察者会接收到

Terminated(actorRef)

消息，消息里包含了被关闭actor的引用。这个观察着或者显式地处理这个消息，或者因为

DeathPactException

而失败。后者在某些情况下是非常有用的，例如一个actor在被观察的actor停止后就不能继续自己的工作了。在我们的情境下，组actor应该在一个设备停止后继续工作，所以我们需要去处理

Terminated(actorRef)

消息。

我们的设备组需要具备以下功能：

1、当新的设备actor被创建时要开始观察它

2、当收到actor停止的消息时，从

Map[String, ActorRef]

中清除它对应的条目

不幸的是，

Terminated

消息只包含

ActorRef

，我们需要获得actor的ID来从map里将其删除。为了有能力删除它，我们需要另一个结构

Map[ActorRef, String]

，这允许我们通过获得的

ActorRef

找出设备ID。

添加了标识功能的actor如下：

class DeviceGroup(groupId: String) extends Actor with ActorLogging {
var deviceIdToActor = Map.empty[String, ActorRef]
var actorToDeviceId = Map.empty[ActorRef, String]

override def preStart(): Unit = log.info("DeviceGroup {} started", groupId)

override def postStop(): Unit = log.info("DeviceGroup {} stopped", groupId)

override def receive: Receive = {
case trackMsg @ RequestTrackDevice(`groupId`, _) ⇒
deviceIdToActor.get(trackMsg.deviceId) match {
case Some(deviceActor) ⇒
deviceActor forward trackMsg
case None ⇒
log.info("Creating device actor for {}", trackMsg.deviceId)
val deviceActor = context.actorOf(Device.props(groupId, trackMsg.deviceId), s"device-${trackMsg.deviceId}")
context.watch(deviceActor)
actorToDeviceId += deviceActor -> trackMsg.deviceId
deviceIdToActor += trackMsg.deviceId -> deviceActor
deviceActor forward trackMsg
}

case RequestTrackDevice(groupId, deviceId) ⇒
log.warning(
"Ignoring TrackDevice request for {}. This actor is responsible for {}.",
groupId, this.groupId
)

case Terminated(deviceActor) ⇒
val deviceId = actorToDeviceId(deviceActor)
log.info("Device actor for {} has been terminated", deviceId)
actorToDeviceId -= deviceActor
deviceIdToActor -= deviceId

}
}

到目前为止，我们没办法获取组actor所跟踪的设备列表，因此我们无法测试我们的这个新功能。为了使其可测，我们添加了一个新的查询功能（

RequestDeviceList(requestId: Long)

消息）：列出当前活动的设备ID：

object DeviceGroup {
def props(groupId: String): Props = Props(new DeviceGroup(groupId))

final case class RequestDeviceList(requestId: Long)
final case class ReplyDeviceList(requestId: Long, ids: Set[String])
}

class DeviceGroup(groupId: String) extends Actor with ActorLogging {
var deviceIdToActor = Map.empty[String, ActorRef]
var actorToDeviceId = Map.empty[ActorRef, String]

override def preStart(): Unit = log.info("DeviceGroup {} started", groupId)

override def postStop(): Unit = log.info("DeviceGroup {} stopped", groupId)

override def receive: Receive = {
case trackMsg @ RequestTrackDevice(`groupId`, _) ⇒
deviceIdToActor.get(trackMsg.deviceId) match {
case Some(deviceActor) ⇒
deviceActor forward trackMsg
case None ⇒
log.info("Creating device actor for {}", trackMsg.deviceId)
val deviceActor = context.actorOf(Device.props(groupId, trackMsg.deviceId), s"device-${trackMsg.deviceId}")
context.watch(deviceActor)
actorToDeviceId += deviceActor -> trackMsg.deviceId
deviceIdToActor += trackMsg.deviceId -> deviceActor
deviceActor forward trackMsg
}

case RequestTrackDevice(groupId, deviceId) ⇒
log.warning(
"Ignoring TrackDevice request for {}. This actor is responsible for {}.",
groupId, this.groupId
)

case RequestDeviceList(requestId) ⇒
sender() ! ReplyDeviceList(requestId, deviceIdToActor.keySet)

case Terminated(deviceActor) ⇒
val deviceId = actorToDeviceId(deviceActor)
log.info("Device actor for {} has been terminated", deviceId)
actorToDeviceId -= deviceActor
deviceIdToActor -= deviceId

}
}

我们几乎已经准备好了测试这个删除功能，但是我们仍需要以下能力：

1、在测试用例里停止我们的设备actor。任何人都可以通过发送一个内置的消息

PoisonPill

从外部停止一个actor。

2、我们需要在设备停止时被通知，我么可以使用死亡观察功能来实现。我们可以轻松使用

TestProbe

拥有的两个消息：使用

watch()

来观察一个特定的actor，使用

expectTerminated

来断言被观察的actor已经被终止。

我们添加两个测试用例。第一个测试用例，我们仅仅测试我们在添加设备后可以获取ID列表。第二个测试用例确保设备ID在设备停止后被适当地移除。

"be able to list active devices" in {
val probe = TestProbe()
val groupActor = system.actorOf(DeviceGroup.props("group"))

groupActor.tell(DeviceManager.RequestTrackDevice("group", "device1"), probe.ref)
probe.expectMsg(DeviceManager.DeviceRegistered)

groupActor.tell(DeviceManager.RequestTrackDevice("group", "device2"), probe.ref)
probe.expectMsg(DeviceManager.DeviceRegistered)

groupActor.tell(DeviceGroup.RequestDeviceList(requestId = 0), probe.ref)
probe.expectMsg(DeviceGroup.ReplyDeviceList(requestId = 0, Set("device1", "device2")))
}

"be able to list active devices after one shuts down" in {
val probe = TestProbe()
val groupActor = system.actorOf(DeviceGroup.props("group"))

groupActor.tell(DeviceManager.RequestTrackDevice("group", "device1"), probe.ref)
probe.expectMsg(DeviceManager.DeviceRegistered)
val toShutDown = probe.lastSender

groupActor.tell(DeviceManager.RequestTrackDevice("group", "device2"), probe.ref)
probe.expectMsg(DeviceManager.DeviceRegistered)

groupActor.tell(DeviceGroup.RequestDeviceList(requestId = 0), probe.ref)
probe.expectMsg(DeviceGroup.ReplyDeviceList(requestId = 0, Set("device1", "device2")))

probe.watch(toShutDown)
toShutDown ! PoisonPill
probe.expectTerminated(toShutDown)

// 因为可能会需要很长时间才能接受到设备actor的停止，因此使用awaitAssert来重试用例
probe.awaitAssert {
groupActor.tell(DeviceGroup.RequestDeviceList(requestId = 1), probe.ref)
probe.expectMsg(DeviceGroup.ReplyDeviceList(requestId = 1, Set("device2")))
}
}

创建设备管理actor

我们继续进入到actor层次结构的下一层，我们需要在源文件

DeviceManager

中创建我们管理组件的入口。这个actor和设备组actor很相似，但是它创建的是设备组actor：

object DeviceManager {
def props(): Props = Props(new DeviceManager)

final case class RequestTrackDevice(groupId: String, deviceId: String)
case object DeviceRegistered
}

class DeviceManager extends Actor with ActorLogging {
var groupIdToActor = Map.empty[String, ActorRef]
var actorToGroupId = Map.empty[ActorRef, String]

override def preStart(): Unit = log.info("DeviceManager started")

override def postStop(): Unit = log.info("DeviceManager stopped")

override def receive = {
case trackMsg @ RequestTrackDevice(groupId, _) ⇒
groupIdToActor.get(groupId) match {
case Some(ref) ⇒
ref forward trackMsg
case None ⇒
log.info("Creating device group actor for {}", groupId)
val groupActor = context.actorOf(DeviceGroup.props(groupId), "group-" + groupId)
context.watch(groupActor)
groupActor forward trackMsg
groupIdToActor += groupId -> groupActor
actorToGroupId += groupActor -> groupId
}

case Terminated(groupActor) ⇒
val groupId = actorToGroupId(groupActor)
log.info("Device group actor for {} has been terminated", groupId)
actorToGroupId -= groupActor
groupIdToActor -= groupId

}

}

因为设备管理的测试用例和我们刚刚写的组actor很像，因此我们把它留给你作为测验。

接下来

我们已经有了一个分级的组件来跟踪设备并且记录测量信息。我们已经看到如何实现不同类型的会话模式，例如：

1、请求-回应（对于临时记录）

2、代理-回应（对于与注册设备）

3、创建-观察-终止（对于创建组和设备actor，并把其作为子actor）

在下一章节，我们会介绍组请求能力，这会建立一个新的分散-聚合会话模式。我们会特别地实现允许用户请求组内所有设备状态的功能。