【AKKA 官方文档翻译】第三部分：与设备Actor一起工作

第三部分：与设备Actor一起工作

akka版本2.5.8

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

在之前的话题中，我们解释了如何在高层次来看待actor系统，即要如何去表示组件，如何安排actor的层次结构。在本节中，我们会看到如何实现其中的设备actor。

如果我们使用对象，我们会将API设计为接口，并拥有一组会被实现类实现的抽象的方法。但是在actor的世界里，协议（protocols）取代了接口。虽然我们不能在编程语言内形式化通用协议，但是我们可以编写它们最基本的元素——消息。因此，我们会从定义我们希望发给设备的消息开始我们的程序。

给设备的消息

设备actor的工作很简单：

1、收集温度测量信息

2、当被查询时，报告最后一次的测量值

然而，在设备启动时不会立刻就获得温度测量信息，因此，我们需要考虑温度测量信息不存在的情况。这也允许我们的actor在没有写模块的时候来测试读模块，因为设备可以简单地报告一个空结果。

从设备获取但前温度的协议很简单，actor需要：

1、等待取当前温度的请求

2、回应这个请求：

①拥有当前的温度数据

②标识当前温度数据还不可用

我们需要两个消息，一个用来请求，一个用来回复。我们的第一次尝试可能如下所示：

final case object ReadTemperature
final case class RespondTemperature(value: Option[Double])

这两条消息貌似涵盖了所有我们所需要的功能，然而，我们选择方法的时候必须要考虑应用程序的分布式特性。虽然actor在JVM本地通信与远程通信的基本机制相同，但是我们需要牢记以下几点：

1、本地信息与远程信息的传输延迟有很大的不同，有些因素，如网络带宽、信息大小都会产生作用。

2、可靠性必须被重视，因为在远程信息传递中会涉及到很多的步骤，这也会增大失败的几率。

3、本地消息仅仅是在JVM内部传递引用，因此不会对消息有很多的限制，但是远程传输可能会限制消息的大小。

另外，在JVM内部传递消息显然是可靠性很高的，但是当actor因为程序员的错误而在处理信息时失败了，那么系统的表现就会和远程网络请求中远程处理消息崩溃一致。尽管这是两个场景，服务一会就会被恢复（actor会被监管者重启，主机会被操作员或监控系统重启），但是个别的请求可能会在故障中丢失。因此，我们要悲观一些，在丢失任何信息的情况下都要保证系统安全

进一步理解协议中的灵活性需求，将有助于我们去考虑Akka消息顺序和消息传递保证。Akka为消息发送提供了以下行为：

1、最多只有一次传递，即不保证送达

2、信息是被每个发送者接收者对来维护的

以下章节将讨论行为中的更多细节：

1、信息传递

2、信息排序

信息传递

消息传递子系统提供的消息传递语义通常分为以下几类：

1、最多传递一次（At-most-once delivery），每个消息被发送零或一次，这意味着信息可能会丢失，但永远不会被重复接收到

2、至少传递一次（At-least-once delivery），每个消息都可能被潜在地发送很多次，直到有一次成功。这意味着信息可能会被重复接收，但永远不会丢失

3、准确地发送一次（Exactly-once delivery），每个消息都被精准地发送给接收者一次，消息不会丢失也不会重复接收

Akka使用第一种行为，它是最节省资源的，并且性能最好。它拥有最小的实现开销，因为可以使用发送即忘（fire-and-forget）
cec2
策略，而不用在发送者内保存发送状态。第二点，也不需要对传输丢失进行计数。这些增加了发送结束后保持状态、发送完毕确认的开销。准确地发送一次信息的方式开销是最大的，由于其很差的性能表现，除了在发送端增加上述所说的开销外，还需要在接收端增加过滤重复消息的机制。

在actor系统中，我们需要确定一个消息被保证的含义，在哪种情况下认为传输已经完成：

1、当消息被送出到网络上时？

2、当消息被接收者主机接收到时？

3、当消息被放到接收者actor的邮箱里时？

4、当消息接收者actor开始处理这个信息时？

5、当消息接受者actor处理完这个消息时

大多数框架和协议声称保证传输，实际上它们提供了类似于4和5的东西。虽然这听起来是合理的，但是实际上真的有用吗？要理解其中的含义，请考虑一个简单的问题：用户尝试下一个订单，并且我们认为一旦它进入了订单数据库，就代表它已经被成功处理了。

如果我们依赖于第五点，即消息被成功处理，那么actor需要尽快在处理完后报告成功状态，这个actor就有义务在订单被提交到它的API后进行校验、处理，然后放入订单数据库。不幸的是，当API被调用后，这些情况可能会发生：

1、主机崩溃

2、反序列化失败

3、校验失败

4、数据库不可访问

5、发生程序错误

这说明传输保证不能被认为是领域级别的保证。我们只想让它在完全处理完订单并将其持久化后报告成功状态。唯一能报告成功状态的实体是应用程序本身，因为只有它了解领域内保证传输需要有哪些需求。没有一个通用的系统可以搞清楚某个特定领域中什么情况才会被认为是成功。

在这个特定的例子中，我们只想在成功写入数据库之后发出成功信号，数据库确认已经安全地将订单存储起来。由于这些原因，Akka将保证程序的责任提升给了应用程序本身，即你必须自己去实现这些。这给了你完全的控制权，让你可以保护你需要保护的内容。现在，让我们考虑下Akka为我们提供的消息排序，以便轻松推理应用程序逻辑。

信息排序

在Akka里对于一个给定的发送接收actor对。直接从A到B的消息不会被无序接收。直接这个词强调这只适用于直接向接收者发动消息，而不包括中间有协调员的情况。

如果：

1、actor

A1

向

A2

发送了信息

M1

，

M2

，

M3

2、actor

A3

向

A2

发送了信息

M4

，

M5

，

M6

这意味着对于Akka消息：

1、

M1

必须在

M2

和

M3

前被发送

2、

M2

必须在

M3

前被发送

3、

M4

必须在

M5

和

M6

前被发送

4、

M5

必须在

M6

前被发送

5、

A2

看到的

A1

和

A3

的信息可能是交错出现的

6、当前我们没有保证传输，所有消息都有可能会被丢弃，比如没有到达

A2

这些保证达到了一个很好的平衡：从一个actor接收到有序的消息使我们可以方便地构建易于推理的系统。另一方面，允许不同actor的消息交错接受给了我们足够的自由度，让我们可以实现高性能的actor系统。

有关传输保证的完整细节，弃权那个参考参考页面。

为设备消息添加灵活性

我们的第一个查询协议是正确的，但是没有考虑分布式应用程序的执行。如果我们想在actor中实现重传（因为请求超时），以便查询设备actor，或者我们想在查询多个actor时关联请求和回复。因此，我们在消息里添加了一个字段，以便请求者可以提供一个ID（我们会在接下来的步骤里把代码添加到应用程序里）：

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

定义设备actor和读取协议

正如我们在Hello World实例里学习到的，每个actor定义了其能接受到的消息种类。我们的设备actor有义务使用相同的ID参数来回应请求，这将看起来如下所示：

import akka.actor.{ Actor, ActorLogging, Props }

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

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 ReadTemperature(id) ⇒
sender() ! RespondTemperature(id, lastTemperatureReading)
}

}

注意代码中的：

1、伴生对象定义了如何创建

Device

actor，期中

props

方法的参数包含设备的ID和所属的组ID，这在之后将会用到。

2、伴生对象包含了我们之前所述的消息的定义。

3、在

Device

类里，

lastTemperatureReading

的值初始化为

None

，并且actor可以简单地将它返回。

测试actor

基于上面的简单actor，我们可以写一个简单的测试用例。在测试代码路径下的

com.lightbend.akka.sample

包里添加

DeviceSpec.scala

文件。（我们使用ScalaTest，你也可以使用其他测试框架）

你可以通过在sbt提示符下运行

test

来运行测试。

"reply with empty reading if no temperature is known" in {
val probe = TestProbe()
val deviceActor = system.actorOf(Device.props("group", "device"))

deviceActor.tell(Device.ReadTemperature(requestId = 42), probe.ref)
val response = probe.expectMsgType[Device.RespondTemperature]
response.requestId should ===(42)
response.value should ===(None)
}

现在当actor接收到传感器的信息时，需要一种方式来改变其温度状态。

添加一个写入协议

写入协议的目的是在接受到包含温度的信息时更新

currentTemperature

字段。同样，我们使用一个简单的消息来定义写入协议，就像这样：

final case class RecordTemperature(value: Double)

然而，这种方式没有考虑让发送者知道温度记录是否被处理，我们已经看到Akka并不保证消息传输，并且把提供消息成功提示留给了应用程序来做。在我们的场景下，我们希望在更新温度之后给发送者一个确认消息。例如：

final case class TemperatureRecorded(requestId: Long)

。就像之前场景中温度的请求和回应一样，添加一个ID字段提供了极大的灵活性。

有读写消息的actor

将读写协议放在一起，设备actor看起来就会像这样：

import akka.actor.{ Actor, ActorLogging, Props }

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 RecordTemperature(id, value) ⇒
log.info("Recorded temperature reading {} with {}", value, id)
lastTemperatureReading = Some(value)
sender() ! TemperatureRecorded(id)

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

我们现在还需要写一个新的测试用例，同时执行读/请求和写/记录：

"reply with latest temperature reading" in {
val probe = TestProbe()
val deviceActor = system.actorOf(Device.props("group", "device"))

deviceActor.tell(Device.RecordTemperature(requestId = 1, 24.0), probe.ref)
probe.expectMsg(Device.TemperatureRecorded(requestId = 1))

deviceActor.tell(Device.ReadTemperature(requestId = 2), probe.ref)
val response1 = probe.expectMsgType[Device.RespondTemperature]
response1.requestId should ===(2)
response1.value should ===(Some(24.0))

deviceActor.tell(Device.RecordTemperature(requestId = 3, 55.0), probe.ref)
probe.expectMsg(Device.TemperatureRecorded(requestId = 3))

deviceActor.tell(Device.ReadTemperature(requestId = 4), probe.ref)
val response2 = probe.expectMsgType[Device.RespondTemperature]
response2.requestId should ===(4)
response2.value should ===(Some(55.0))
}

接下来

到目前为止，我们已经开始设计我们的整体架构，并且我们编写了与领域直接对应的第一个actor。我们之后需要创建一个用来维护设备组和设备actor的组件。