Akka 和 Java 内存模型

使用 LightBend 平台（包括 Scala 和 Akka）的一个主要好处是简化了并发软件的编写过程。本文讨论了 LightBend 平台，特别是 Akka 如何在并发应用程序中处理共享内存。

Java 内存模型

在 Java 5 之前，Java 内存模型（JMM）是定义有问题的。当多个线程访问共享内存时，可能会得到各种奇怪的结果，例如：

  • 线程看不到其他线程写入的值：可见性问题。
  • 由于没有按预期的顺序执行指令而导致的观察其他线程发生“不可能”的行为：指令重新排序问题。

随着 Java 5 中 JSR 133 的实现，许多问题得到了解决。JMM 是基于“先于发生（happens-before）”关系的一组规则，它约束一个内存访问必须发生在另一个内存访问之前的时间。这些规则的两个例子是：

  • 监视器锁规则：在每次后续获取同一个锁之前，都会释放一个锁。
  • volatile变量规则：volatile变量的写入发生在同一个volatile变量的每次后续读取之前。

虽然 JMM 看起来很复杂，但是规范试图在易用性和编写性能、可扩展并发数据结构的能力之间找到一个平衡点。

Actors 和 Java 内存模型

通过 Akka 中的 Actor 实现，多个线程可以通过两种方式在共享内存上执行操作：

  • 如果消息发送给某个 Actor（例如由另一个 Actor）。在大多数情况下，消息是不可变的，但是如果该消息不是正确构造的不可变对象，没有“先于发生”规则，则接收者可能会看到部分初始化的数据结构，甚至可能会看到空气稀薄的值（longs/doubles）。
  • 如果 Actor 在处理消息时更改其内部状态，并在稍后处理另一条消息时访问该状态。重要的是要认识到，对于 Actor 模型，你不能保证同一线程将对不同的消息执行相同的 Actor。

为了防止 Actor 出现可见性和重新排序问题，Akka 保证以下两条“先于发生”规则：

  • Actor 发送规则：向 Actor 发送消息的过程发生在同一个 Actor 接收消息之前。
  • Actor 后续处理规则：一条消息的处理发生在同一个 Actor 处理下一条消息之前。

注释：在外行术语中，这意味着当 Actor 处理下一条消息时，Actor 内部字段的更改是可见的。因此，Actor 中的字段不必是volatile或equivalent的。

这两个规则仅适用于同一个 Actor 实例，如果使用不同的 Actor，则这两个规则无效。

Futures 和 Java 存储模型

Future的“先于发生”调用任何注册到它的回调被执行之前。

我们建议不要关闭非final字段（Java 中的final和 Scala 中的val），如果选择关闭非final字段，则必须标记volatile，以便字段的当前值对回调可见。

如果关闭引用，还必须确保引用的实例是线程安全的。我们强烈建议远离使用锁的对象，因为它可能会导致性能问题，在最坏的情况下还会导致死锁。这就是同步的危险。

Actors 和共享可变状态

由于 Akka 在 JVM 上运行，所以仍然需要遵循一些规则。

  • 关闭内部 Actor 状态并将其暴露给其他线程
  1. import akka.actor.{ Actor, ActorRef }
  2. import akka.pattern.ask
  3. import akka.util.Timeout
  4. import scala.concurrent.Future
  5. import scala.concurrent.duration._
  6. import scala.language.postfixOps
  7. import scala.collection.mutable
  9. case class Message(msg: String)
  11. class EchoActor extends Actor {
  12.   def receive = {
  13.     case msg ⇒ sender() ! msg
  14.   }
  15. }
  17. class CleanUpActor extends Actor {
  18.   def receive = {
  19.     case set: mutable.Set[_] ⇒ set.clear()
  20.   }
  21. }
  23. class MyActor(echoActor: ActorRef, cleanUpActor: ActorRef) extends Actor {
  24.   var state = ""
  25.   val mySet = mutable.Set[String]()
  27.   def expensiveCalculation(actorRef: ActorRef): String = {
  28.     // this is a very costly operation
  29.     "Meaning of life is 42"
  30.   }
  32.   def expensiveCalculation(): String = {
  33.     // this is a very costly operation
  34.     "Meaning of life is 42"
  35.   }
  37.   def receive = {
  38.     case _ ⇒
  39.       implicit val ec = context.dispatcher
  40.       implicit val timeout = Timeout(5 seconds) // needed for `?` below
  42.       // Example of incorrect approach
  43.       // Very bad: shared mutable state will cause your
  44.       // application to break in weird ways
  45.       Future { state = "This will race" }
  46.       ((echoActor ? Message("With this other one")).mapTo[Message])
  47.         .foreach { received ⇒ state = received.msg }
  49.       // Very bad: shared mutable object allows
  50.       // the other actor to mutate your own state,
  51.       // or worse, you might get weird race conditions
  52.       cleanUpActor ! mySet
  54.       // Very bad: "sender" changes for every message,
  55.       // shared mutable state bug
  56.       Future { expensiveCalculation(sender()) }
  58.       // Example of correct approach
  59.       // Completely safe: "self" is OK to close over
  60.       // and it's an ActorRef, which is thread-safe
  61.       Future { expensiveCalculation() } foreach { self ! _ }
  63.       // Completely safe: we close over a fixed value
  64.       // and it's an ActorRef, which is thread-safe
  65.       val currentSender = sender()
  66.       Future { expensiveCalculation(currentSender) }
  67.   }
  68. }
  • 消息应该是不可变的，这是为了避免共享可变状态陷阱。

英文原文链接：Akka and the Java Memory Model.