JVM 并发性: 使用 Akka 执行异步操作

本 系列 中以前的文章介绍了如何通过以下方式实现并发性：

• 并行地在多个数据集上执行相同的操作（就像 Java 8 流一样）
• 显式地将计算构建成异步执行某些操作，然后将结果组合在一起（就像 future 一样）。

这两种方法都是实现并发性的不错方式，但是您必须将它们明确地设计到应用程序中。

在本文和接下来的几篇文章中，我将着重介绍一种不同的并发性实现方法，该方法基于一种特定的程序结构，与显式编码方法不同。这种程序结构就是 actor 模型 。您将了解如何使用 actor 模型的  Akka 实现。（Akka 是一个构建并发和分布式 JVM 应用程序的工具包和运行时。）请参阅  参考资料 ，获取本文完整示例代码的链接。

Actor 模型基础知识

用于并发计算的 actor 模型基于各种称为 actor 的原语来构建系统。Actor 执行操作来响应称为 消息 的输入。这些操作包括更改 actor 自己的内部状态，以及发出其他消息和创建其他 actor。所有消息都是异步交付的，因此将消息发送方与接收方分开。正是由于这种分离，导致 actor 系统具有内在的并发性：可以不受限制地并行执行任何拥有输入消息的 actor。

在 Akka 术语中，actor 看起来就像是某种通过消息进行交互的行为神经束。像真实世界的演员一样，Akka actor 也需要一定程度的隐私。您不能直接将消息发送给 Akka actor。相反，需要将消息发送给等同于邮政信箱的 actor 引用 。然后通过该引用将传入的消息路由到 actor 的邮箱，以后再传送给 actor。Akka actor 甚至要求所有传入的消息都是无菌的（或者在 JVM 术语中叫做 不可变的 ），以免受到其他 actor 的污染。

与一些真实世界中演员的需求不同，Akka 中由于某种原因而存在一些看似强制要求的限制。使用 actor 的引用可阻止交换消息以外的任何交互，这些交互可能破坏 actor 模型核心上的解耦本质。Actor 在执行上是单线程的（不超过 1 个线程执行一个特定的 actor 实例），所以邮箱充当着一个缓冲器，在处理消息前会一直保存这些消息。消息的不可变性（由于 JVM 的限制，目前未由 Akka 强制执行，但这是一项既定的要求）意味着根本无需担心可能影响 actor 之间各种共享的数据的同步问题；如果只有共享的数据是不可变的，那么根本不需要同步。

开始实现

现在您已大体了解了 actor 模型和 Akka 细节，是时候看些代码了。使用 hello 作为编码示例司空见惯，但它确实能够帮助用户快速、轻松地理解一种语言或系统。清单 1 显示了 Scala 中的一个 Akka 版本。

清单 1. 简单的 Scala hello

import akka.actor._ import akka.util._  /** Simple hello from an actor in Scala. */ object Hello1 extends App {      val system = ActorSystem("actor-demo-scala")   val hello = system.actorOf(Props[Hello])   hello ! "Bob"   Thread sleep 1000   system shutdown      class Hello extends Actor {     def receive = {       case name: String => println(s"Hello $name")     }   } }

清单 1 中的代码分为两个单独的代码段，它们都包含在 Hello1 应用程序项目中。第一个代码段是 Akka 应用程序基础架构，它：

1. 创建一个 actor 系统（ ActorSystem(...)  行）。
2. 在系统内创建一个 actor（ system.actorOf(...)  行，它为所创建的 actor 返回一个 actor 引用）。
3. 使用 actor 引用向 actor 发送消息（ hello !"Bob"  行）。
4. 等待一秒钟，然后关闭 actor 系统（ system shutdown  行）。

system.actorOf(Props[Hello]) 调用是创建 actor 实例的推荐方式，它使用了专门用于  Hello actor 类型的配置属性。对于这个简单的 actor（扮演一个小角色，只有一句台词），没有配置信息，所以  Props 对象没有参数。如果想在您的 actor 上设置某种配置，可专门为该 actor 定义一个其中包含了所有必要信息的  Props 类。（后面的示例会展示如何操作。）

hello !"Bob" 语句将一条消息（在本例中为字符串  Bob ）发送给已创建的 actor。 ! 运算符是 Akka 中表示将一条消息发送到 actor 的便捷方式，采用了触发即忘的模式。如果不喜欢这种特殊的运算符风格，可使用  tell() 方法实现相同的功能。

第二段代码是 Hello actor 定义，以  class Hello extends Actor 开头。这个特定的 actor 定义非常简单。它定义必需的（对于所有 actor）局部函数  receive ，该函数实现了传入消息的处理方式。（ receive 是一个 局部函数 ，因为仅为一些输入定义了它 — 在本例中，仅为  String 消息输入定义了该函数。）为这个 actor 所实现的处理方法是，只要收到一条  String 消息，就使用该消息值打印一条问候语。

Java 中的 Hello

清单 2 给出了清单 1 中的 Akka Hello 在普通 Java 中的表示。

清单 2. Java 中的 Hello

import akka.actor.*; public class Hello1 {  public static void main(String[] args) {   ActorSystem system = ActorSystem.create("actor-demo-java");   ActorRef hello = system.actorOf(Props.create(Hello.class));   hello.tell("Bob", ActorRef.noSender());   try {    Thread.sleep(1000);   } catch (InterruptedException e) { /* ignore */ }   system.shutdown();  }  private static class Hello extends UntypedActor {   public void onReceive(Object message) throws Exception {    if (message instanceof String) {     System.out.println("Hello " + message);    }   }  } } 

清单 3 显示了来自包含 lambda 的 Java 8 的 actor 定义，以及 lambda 支持的 ReceiveBuilder 类所需要的导入。清单 3 或许更加紧凑，但与清单 2 大同小异。

清单 3. Java 8 的 Akka Hello 版本

import akka.japi.pf.ReceiveBuilder; ...  private static class Hello extends AbstractActor {   public Hello() {    receive(ReceiveBuilder.     match(String.class, s -> { System.out.println("Hello " + s); }).     build());   }  } 

与清单 2 相比，清单 3 中的 Java 8 代码使用了一个不同的基类 — AbstractActor 代替  UntypedActor — 而且还使用了一种不同的方式来定义消息处理方案。 ReceiveBuilder 类允许您使用 lambda 表达式来定义消息的处理方式，并采用了类似 Scala 的匹配语法。如果您主要在 Scala 中进行开发工作，此技术可能有助于让 Java Akka 代码看起来更简洁，但使用 Java 8 特定版本的好处似乎有些微不足道。

为什么还要等待？

在主应用程序代码中，将消息发送到 actor 之后，会有一次 Thread sleep 1000 形式的等待，然后才会关闭系统。您可能想知道为什么需要等待。毕竟，消息很容易处理；难道消息没有立即传到 actor，在  hello !"Bob" 语句完成时还在处理当中？

这个问题的答案很简单：“不是”。Akka actor 是异步运行的，所以即使目标 actor 与发送方 actor 位于相同的 JVM 中，目标 actor 也绝不会立即开始执行。相反，处理该消息的线程会将消息添加到目标 actor 的邮箱中。将消息添加到邮箱中会触发一个线程，以便从邮箱获取该消息并调用 actor 的 receive 方法来处理。但从邮箱获取消息的线程通常不同于将消息添加到邮箱的线程。

消息传送时间和保证

“为什么还要等待？” 这一问题的简短答案的背后是一种更深入的原理。Akka 支持 actor 远程通信且具有位置透明性，意味着您的代码没有任何直接的方式来了解一个特定的 actor 是位于同一 JVM 中，还是在系统外的云中某处运行。但这两种情况在实际操作中显然具有完全不同的特征。

一个差别与消息丢失有关。Akka 无法保证消息将被传送到目的地，熟悉消息传递系统（用于连接应用程序）的开发人员可能对此很吃惊。这种无保证传送背后的哲学原理是 Akka 的核心原理之一：针对失败而设计。作为一种有意为之的过度简化，可以认为传送保证为消息传输系统添加了很高的复杂性，而且这些更复杂的系统有时无法按预期运行，而应用程序代码还必须涉及恢复操作。这种原理在应用程序代码始终自行处理传送失败情况时很有意义，能够让消息传送系统保持简单。

Akka 可以 保证消息最多传送一次，而且绝不会无序地收到从一个 actor 实例发送到另一个 actor 实例的消息。但后者仅适用于特定的 actor 对，二者没有联系。如果 actor A 将消息发送给 actor B，这些消息绝不会被打乱顺序。如果 actor A 将消息发送给 actor C，情况也是如此。但是，如果 actor B 也将消息发送给 actor C（例如将来自 A 的消息转发给 C），B 的消息相对于来自 A 的消息而言可能是乱序的。

在 清单 1 的代码中，消息丢失的概率非常低，因为代码在单个 JVM 中运行，不会生成过多的消息负载。（过多的消息负载可能导致消息丢失。如果 Akka 没有空间来存储消息，例如它没有备用方案，那么只能丢弃消息。）但清单 1 代码的结构仍未对消息传送事件做出任何假设，而且允许 actor 系统执行异步操作。

Actor 和状态

Akka 的 actor 模型很灵活，支持所有类型的 actor。可以使用没有状态信息的 actor（就像 Hello1 示例中一样），但这些 actor 可能等效于方法调用。添加状态信息可实现更为灵活的 actor 函数。

清单 1 提供了一个完整的（但很普通）actor 系统示例 — 但拥有一个始终执行同一工作的 actor。每位演员都讨厌反复重复同一句话，所以清单 4 向 actor 添加了一些状态信息，使工作变得更有趣。

清单 4. Polyglot Scala hello

object Hello2 extends App {      case class Greeting(greet: String)   case class Greet(name: String)      val system = ActorSystem("actor-demo-scala")   val hello = system.actorOf(Props[Hello], "hello")   hello ! Greeting("Hello")   hello ! Greet("Bob")   hello ! Greet("Alice")   hello ! Greeting("Hola")   hello ! Greet("Alice")   hello ! Greet("Bob")   Thread sleep 1000   system shutdown      class Hello extends Actor {     var greeting = ""     def receive = {       case Greeting(greet) => greeting = greet       case Greet(name) => println(s"$greeting $name")     }   } }

清单 4 中的 actor 知道如何处理两种不同类型的消息，这些消息在清单的开头附近定义： Greeting 消息和  Greet 消息，每条消息都包装了一个字符串值。修改后的  Hello actor 收到  Greeting 消息时，会将所包装的字符串保存为  greeting 值。收到  Greet 消息时，则将已保存的 greeting 值与  Greet 字符串组合起来，形成最终的消息。下面在运行此应用程序时，我们可以看到在控制台中打印出的消息（但消息不一定是按此顺序出现的，因为 actor 执行顺序是不确定的）：

Hello Bob Hello Alice Hola Alice Hola Bob

清单 4 中并没有太多的新代码，所以我没有提供其 Java 版本。您可在代码下载内容中找到它们（参见 参考资料 ），名为 com.sosnoski.concur.article5java.Hello2 和  com.sosnoski.concur.article5java8.Hello2 。

属性和交互

真正的 actor 系统会使用多个 actor 来完成工作，它们彼此发送消息来进行交互。并且常常需要为这些 actor 提供配置信息，以准备履行其具体的职责。清单 5 基于 Hello 示例中使用的技术，展示了简化版的 actor 配置和交互。

清单 5. Actor 属性和交互

object Hello3 extends App {    import Greeter._   val system = ActorSystem("actor-demo-scala")   val bob = system.actorOf(props("Bob", "Howya doing"))   val alice = system.actorOf(props("Alice", "Happy to meet you"))   bob ! Greet(alice)   alice ! Greet(bob)   Thread sleep 1000   system shutdown    object Greeter {     case class Greet(peer: ActorRef)     case object AskName     case class TellName(name: String)     def props(name: String, greeting: String) = Props(new Greeter(name, greeting))   }    class Greeter(myName: String, greeting: String) extends Actor {     import Greeter._     def receive = {       case Greet(peer) => peer ! AskName       case AskName => sender ! TellName(myName)       case TellName(name) => println(s"$greeting, $name")     }   } }

清单 5 在领导角色中包含了一个新的 actor，即 Greeter actor。 Greeter 在  Hello2 示例的基础上更进了一步，包含：

• 所传递的属性，目的是配置  Greeter  实例
• 定义了配置属性和消息的 Scala 配套对象（如果您有 Java 工作背景，可将这个配套对象视为与 actor 类同名的静态 helper 类）
• 在  Greeter  actor 的实例间发送的消息

此代码生成的输出很简单：

Howya doing, Alice Happy to meet you, Bob

如果尝试运行该代码几次，可能会看到这些行的顺序是相反的。这种排序是 Akka actor 系统动态本质的另一个例子，其中处理各个消息时的顺序是不确定的（但包含我在 “ 消息传送时间和保证 ” 中讨论的几个重要例外）。

Java 中的 Greeter

清单 6 显示了清单 5 中的 Akka Greeter 代码的普通 Java 版本。

清单 6. Java 中的 Greeter

public class Hello3 {  public static void main(String[] args) {   ActorSystem system = ActorSystem.create("actor-demo-java");   ActorRef bob = system.actorOf(Greeter.props("Bob", "Howya doing"));   ActorRef alice = system.actorOf(Greeter.props("Alice", "Happy to meet you"));   bob.tell(new Greet(alice), ActorRef.noSender());   alice.tell(new Greet(bob), ActorRef.noSender());   try {    Thread.sleep(1000);   } catch (InterruptedException e) { /* ignore */ }   system.shutdown();  }  // messages  private static class Greet {   public final ActorRef target;   public Greet(ActorRef actor) {    target = actor;   }  }  private static Object AskName = new Object();  private static class TellName {   public final String name;   public TellName(String name) {    this.name = name;   }  }  // actor implementation  private static class Greeter extends UntypedActor {   private final String myName;   private final String greeting;   Greeter(String name, String greeting) {    myName = name;    this.greeting = greeting;   }   public static Props props(String name, String greeting) {    return Props.create(Greeter.class, name, greeting);     }   public void onReceive(Object message) throws Exception {    if (message instanceof Greet) {     ((Greet)message).target.tell(AskName, self());    } else if (message == AskName) {     sender().tell(new TellName(myName), self());    } else if (message instanceof TellName) {     System.out.println(greeting + ", " + ((TellName)message).name);    }   }  } } 

清单 7 显示了包含 lambda 的 Java 8 版本。同样，此版本在消息处理的实现方面要更为紧凑，但其他方面都是相同的。

清单 7. Java 8 版本

import akka.japi.pf.ReceiveBuilder; ...  private static class Greeter extends AbstractActor {   private final String myName;   private final String greeting;   Greeter(String name, String greeting) {    myName = name;    this.greeting = greeting;    receive(ReceiveBuilder.     match(Greet.class, g -> { g.target.tell(AskName, self()); }).     matchEquals(AskName, a -> { sender().tell(new TellName(myName), self()); }).     match(TellName.class, t -> { System.out.println(greeting + ", " + t.name); }).     build());   }   public static Props props(String name, String greeting) {    return Props.create(Greeter.class, name, greeting);     }  } 

传递属性

Akka 使用 Props 对象将各种配置属性传递给 actor。每个  Props 实例包装 actor 类所需的构造函数参数的一个副本，以及对该类的引用。可通过两种方式将此信息传递给  Props 构造函数。 清单 5 中的示例将 actor 的构造函数作为一个名称传递 (pass-by-name) 参数传递给  Props 构造函数。注意，此方式不会直接调用构造函数并传递结果；它传递构造函数调用（如果您有 Java 工作背景，可能觉得这很陌生）。

将 actor 配置传递给 Props 构造函数的另一种方法是，提供 actor 的类作为第一个参数，将 actor 的构造函数参数作为剩余的参数。对于  清单 5 中的示例，这种调用形式为  Props(classOf[Greeter], name, greeting) 。

无论使用哪种形式的 Props 构造函数，传递给新 actor 的值都需要可序列化，以便在必要时通过网络将  Props 发送到可运行该 actor 实例的任何地方。对于名称传递构造函数调用的情况，就像  清单 5 中的用法，需要将调用发送出 JVM 时，会序列化调用的闭包。

在 Scala 代码中创建 Props 对象的 Akka 建议做法是：在一个配套对象中定义工厂方法，就像  清单 5 中所做的那样。对  Props 使用名称传递构造函数调用方法时，此技术可阻止任何问题意外地关闭对 actor 对象的  this 引用。配套对象也是定义 actor 将接收的各种消息的不错地方，这样，所有关联的信息都位于同一位置。对于 Java actor，也可在 actor 类中使用静态构造函数方法，如  清单 6 中所用的方法。

发送消息的 Actor

清单 5 中的每个  Greeter actor 都配置了一个名称和一句问候语，但将问候语告知另一个 actor 时，首先要找到另一个 actor 的名称。 Greeter actor 通过向另一个 actor 发送一条单独的消息来完成此任务： AskName 消息。 AskName 消息本身不含任何信息，但收到它的  Greeter 实例知道应使用一个包含  TellName 发送方名称的  TellName 消息作为响应。当第一个  Greeter 收到所返回的  TellName 消息时，它打印出自己的问候语。

发送给 actor 的每个消息都包含由 Akka 提供的一些附加信息，最特别的是消息发送方的 ActorRef 。您可在消息处理过程中的任何时刻，通过调用在 actor 基类上定义的  sender() 方法来访问这些发送方的信息。 Greeter actor 在处理  AskName 消息的过程中会使用发送方引用，以便将  TellName 响应发送给正确的 actor。

Akka 允许您代表另一个 actor 发送消息（一种良性的身份盗窃形式），以便收到该消息的 actor 将另一个 actor 视为发送方。这是在 actor 系统中经常使用的一个有用特性，尤其是对于请求-响应类型的消息交换，因为此时您希望将响应传送到不同于发出请求的 actor 的某个地方。actor 外部的应用程序代码所发出的消息，默认将使用名为 deadletter actor 的特殊 Akka 作为发送方。任何时候无法将消息传送给 actor 时，也可使用 deadletter actor，这为用户提供了一种便捷的方式，在 actor 系统中通过打开合适的日志（我将在下一期中介绍）来跟踪无法传送的消息。

设置 actor 的类型

您可能注意到了，示例的消息序列中没有任何类型的信息来明确表明消息的目标是 Greeter 实例。Akka actor 及其交换的消息一般都属于这种情况。甚至用于表示消息目标 actor 的  ActorRef 也是无类型的。

编写无类型的 actor 系统有着实际的优势。您 可以 定义 actor 类型（比如通过它们可处理的一组消息），但这么做有误导性。在 Akka 中，actor 可以改变它们的行为（下一期会更详细地介绍此内容），所以不同的消息集可能适合不同的 actor 状态。类型也可能妨碍我们合理地简化 actor 模型，因为系统将所有 actor 视为至少拥有处理任何消息的潜力。

但是 Akka 仍然支持类型化 actor，以防您确实想要使用这种方法。这种支持在连接 actor 和非 actor 节点时最有用。您可定义一个接口，非 actor 节点使用该接口与 actor 进行通信，使 actor 看起来更像是正常的程序组件。对于大部分操作，这样做所带来的麻烦太多，可能不值得去做，但考虑到从 actor 系统外部直接将消息发送给 actor 的简单性（从目前为止的任何示例应用程序中可以看到，非 actor 代码可以发送消息），有这个选项也很不错。

消息和可变性

Akka 希望您肯定不会意外地在 actor 之间共享可变的数据。如果共享可变的数据，结果会非常糟 — 比不上在对抗幽灵时穿过自己的质子束（如果您不太熟悉，参见电影 做鬼敢死队 ），但仍然很糟。共享可变数据的问题在于，actor 在单独的线程中运行。如果在 actor 之间共享可变数据，则无法在运行 actor 的线程之间进行协调，所以各个线程不会看到其他线程正在做什么，并且可能通过多种不同的方式对彼此造成破坏。如果正在运行分布式系统，问题会更严重，每个 actor 都将拥有自己的可变数据副本。

所以消息必须是不可变的，而且不仅仅是在表面层面上。如果消息数据中包含任何对象，这些对象必须也是不可变的，依此类推，一直到消息所引用的所有对象。Akka 目前不能强制实施此要求，但 Akka 开发人员希望在将来的某个时刻能强制实施这些限制。如果您希望自己代码在未来的 Akka 版本中仍可使用，那么现在必须留意这一要求。

询问与告诉

清单 5 中的代码使用标准  tell 操作来发送消息。在 Akka 中，也可使用  ask 消息模式作为一种辅助性操作。 ask 操作（由  ? 运算符或使用 ask 函数表示）发送一条包含  Future 的消息作为响应。在清单 8 中，我们重建了清单 5 中的代码，使用  ask 来代替  tell 。

清单 8. 使用 ask

import scala.concurrent.duration._ import akka.actor._ import akka.util._ import akka.pattern.ask  object Hello4 extends App {    import Greeter._   val system = ActorSystem("actor-demo-scala")   val bob = system.actorOf(props("Bob", "Howya doing"))   val alice = system.actorOf(props("Alice", "Happy to meet you"))   bob ! Greet(alice)   alice ! Greet(bob)   Thread sleep 1000   system shutdown    object Greeter {     case class Greet(peer: ActorRef)     case object AskName     def props(name: String, greeting: String) = Props(new Greeter(name, greeting))   }    class Greeter(myName: String, greeting: String) extends Actor {     import Greeter._     import system.dispatcher     implicit val timeout = Timeout(5 seconds)     def receive = {       case Greet(peer) => {         val futureName = peer ? AskName         futureName.foreach { name => println(s"$greeting, $name") }       }       case AskName => sender ! myName     }   } }

在清单 8 的代码中， TellName 消息已被替换为  ask 。 ask 操作返回的 future 的类型为  Future[Any] ，因为编译器对要返回的结果一无所知。当 future 完成时， foreach 使用  import system.dispatcher 语句所定义的隐式调度器来执行  println 。如果 future 未完成且在允许的超时（另一个隐式值，在本例中定义为 5 秒）内提供了响应消息，它会完成并抛出超时异常。

在幕后， ask 模式创建一个特殊的一次性 actor 在消息交换中充当中介。该中介会收到一个  Promise 和要发送的消息，以及目标 actor 引用。它发送消息，然后等待期望的响应消息。收到响应后，它会履行承诺并完成最初的 actor 所使用的 future。

使用 ask 方法有一些限制。具体来讲，要避免公开 actor 状态（可能导致线程问题），必须确保您未在 future 完成时所执行的代码中使用来自该 actor 的任何可变状态。在实际情况中，为在 actor 之间发送的消息使用  tell 模式通常要更容易。ask 模式更有用的一种情况是，应用程序代码需要从 actor（无论是否具有类型）获取响应时（比如启动 actor 系统和创建初始 actor 的主程序）。

小角色

ask 模式所创建的一次性 actor 是在使用 Akka 时要记住的一种出色设计原则。通常您希望构造您的 actor 系统，以便中间处理步骤是由为这种特定的用途而设计的特殊 actor 所执行的。一个常见的例子是，需要在进入下一处理阶段之前合并不同的异步结果。如果为不同的结果使用消息，您可让一个 actor 来收集各个结果，直到所有结果都准备好，然后触发下一阶段的操作。这基本上就是 ask 模式所用的一般的一次性 actor。

Akka actor 是轻型的（每个 actor 实例大约 300 到 400 字节，无论 actor 类使用哪种存储都是如此），所以您可安全地设计程序结构，在适当的时候使用多个 actor。使用专用的 actor 有助于保持代码简单且易于理解，这也是编写并发程序与编写顺序程序相比的一个优势。在对您明确处理异步操作有所帮助时，应毫不犹豫地向设计中引入新的 actor。

补充几句

Akka 是一个强大的系统，但 Akka 和 actor 模型通常都需要一种与直观的过程代码不同的编程风格。对于过程代码，程序结构中的所有调用都是确定的，并且您可查看程序的整个调用树。在 actor 模型中，消息是被乐观地触发的，无法保证它们将会送达，而且常常很难确定事件发生的顺序。actor 模型的好处是，这是一种构建高并发性和可伸缩性应用程序的轻松方式，我在后面的几期中会再次介绍此主题。

希望本文能够让您足够清楚地了解 Akka，激起您进一步探索该内容的欲望。下一次我将更深入地介绍 actor 系统和 actor 交互，包括如何轻松地跟踪系统中各 actor 之间的交互。