面向 Java 开发人员的 Scala 指南: 深入了解 Scala 并发性

http://www.ibm.com/developerworks/cn/java/j-scala04109.html

http://www.ibm.com/developerworks/cn/java/j-scala02049.html

在前一篇文章中，我讨论了构建并发代码的重要性（无论是否是 Scala 代码），还讨论了在编写并发代码时开发人员面对的一些问题，包括不要锁住太多东西、不要锁住太少东西、避免死锁、避免生成太多线程等等。

这些理论问题太沉闷了。为了避免读者觉得失望，我与您一起研究了 Scala 的一些并发构造，首先是在 Scala 中直接使用 Java 语言的并发库的基本方法，然后讨论 Scala API 中的 MailBox 类型。尽管这两种方法都是可行的，但是它们并不是 Scala 实现并发性的主要机制。

真正提供并发性的是 Scala 的 actor。

什么是 “actor”？

“actor” 实现在称为 actor 的执行实体之间使用消息传递进行协作（注意，这里有意避免使用 “进程”、“线程” 或 “机器” 等词汇）。尽管它听起来与 RPC 机制有点儿相似，但是它们是有区别的。RPC 调用（比如 Java RMI 调用）会在调用者端阻塞，直到服务器端完成处理并发送回某种响应（返回值或异常），而消息传递方法不会阻塞调用者，因此可以巧妙地避免死锁。

仅仅传递消息并不能避免错误的并发代码的所有问题。另外，这种方法还有助于使用 “不共享任何东西” 编程风格，也就是说不同的 actor 并不访问共享的数据结构（这有助于促进封装 actor，无论 actor 是 JVM 本地的，还是位于其他地方） — 这样就完全不需要同步了。毕竟，如果不共享任何东西，并发执行就不涉及任何需要同步的东西。

这不算是对 actor 模型的正规描述，而且毫无疑问，具有更正规的计算机科学背景的人会找到各种更严谨的描述方法，能够描述 actor 的所有细节。但是对于本文来说，这个描述已经够了。在网上可以找到更详细更正规的描述，还有一些学术文章详细讨论了 actor 背后的概念（请您自己决定是否要深入学习这些概念）。现在，我们来看看 Scala actors API。

Scala actor

使用 actor 根本不困难，只需使用 Actor 类的 actor 方法创建一个 actor，见清单 1：

清单 1. 开拍！

import scala.actors._, Actor._package com.tedneward.scalaexamples.scala.V4
{object Actor1{def main(args : Array[String]) ={val badActor =actor{receive{case msg => System.out.println(msg)}}badActor ! "Do ya feel lucky, punk?"}}
}

这里同时做了两件事。

首先，我们从 Scala Actors 库的包中导入了这个库，然后从库中直接导入了 Actor 类的成员；第二步并不是完全必要的，因为在后面的代码中可以使用 Actor.actor 替代 actor，但是这么做能够表明 actor 是语言的内置结构并（在一定程度上）提高代码的可读性。

下一步是使用 actor 方法创建 actor 本身，这个方法通过参数接收一个代码块。在这里，代码块执行一个简单的 receive（稍后讨论）。结果是一个 actor，它被存储在一个值引用中，供以后使用。

请记住，除了消息之外，actor 不使用其他通信方法。使用 ! 的代码行实际上是一个向 badActor 发送消息的方法，这可能不太直观。Actor 内部还包含另一个 MailBox 元素（已讨论）；! 方法接收传递过来的参数（在这里是一个字符串），把它发送给邮箱，然后立即返回。

消息交付给 actor 之后，actor 通过调用它的 receive 方法来处理消息；这个方法从邮箱中取出第一个可用的消息，把它交付给一个模式匹配块。注意，因为这里没有指定模式匹配的类型，所以任何消息都是匹配的，而且消息被绑定到 msg 名称（为了打印它）。

一定要注意一点：对于可以发送的类型，没有任何限制 — 不一定要像前面的示例那样发送字符串。实际上，基于 actor 的设计常常使用 Scala case 类携带实际消息本身，这样就可以根据 case 类的参数/成员的类型提供隐式的 “命令” 或 “动作”，或者向动作提供数据。

例如，假设希望 actor 用两个不同的动作来响应发送的消息；新的实现可能与清单 2 相似：

清单 2. 嗨，我是导演！

  object Actor2{case class Speak(line : String);case class Gesture(bodyPart : String, action : String);case class NegotiateNewContract;def main(args : Array[String]) ={val badActor =actor{receive{case NegotiateNewContract =>System.out.println("I won't do it for less than $1 million!")case Speak(line) =>System.out.println(line)case Gesture(bodyPart, action) =>System.out.println("(" + action + "s " + bodyPart + ")")case _ =>System.out.println("Huh? I'll be in my trailer.")}}badActor ! NegotiateNewContractbadActor ! Speak("Do ya feel lucky, punk?")badActor ! Gesture("face", "grimaces")badActor ! Speak("Well, do ya?")}}

到目前为止，看起来似乎没问题，但是在运行时，只协商了新合同；在此之后，JVM 终止了。初看上去，似乎是生成的线程无法足够快地响应消息，但是要记住在 actor 模型中并不处理线程，只处理消息传递。这里的问题其实非常简单：一次接收使用一个消息，所以无论队列中有多少个消息正在等待处理都无所谓，因为只有一次接收，所以只交付一个消息。

纠正这个问题需要对代码做以下修改，见清单 3：

把 receive 块放在一个接近无限的循环中。
创建一个新的 case 类来表示什么时候处理全部完成了。

清单 3. 现在我是一个更好的导演！

  object Actor2{case class Speak(line : String);case class Gesture(bodyPart : String, action : String);case class NegotiateNewContract;case class ThatsAWrap;def main(args : Array[String]) ={val badActor =actor{var done = falsewhile (! done){receive{case NegotiateNewContract =>System.out.println("I won't do it for less than $1 million!")case Speak(line) =>System.out.println(line)case Gesture(bodyPart, action) =>System.out.println("(" + action + "s " + bodyPart + ")")case ThatsAWrap =>System.out.println("Great cast party, everybody! See ya!")done = truecase _ =>System.out.println("Huh? I'll be in my trailer.")}}}badActor ! NegotiateNewContractbadActor ! Speak("Do ya feel lucky, punk?")badActor ! Gesture("face", "grimaces")badActor ! Speak("Well, do ya?")badActor ! ThatsAWrap}}

这下行了！使用 Scala actor 就这么容易。

并发地执行动作

上面的代码没有反映出并发性 — 到目前为止给出的代码更像是另一种异步的方法调用形式，您看不出区别。（从技术上说，在第二个示例中引入接近无限循环之前的代码中，可以猜出有一定的并发性存在，但这只是偶然的证据，不是明确的证明）。

为了证明在幕后确实有多个线程存在，我们深入研究一下前一个示例：

清单 4. 我要拍特写了

  object Actor3{case class Speak(line : String);case class Gesture(bodyPart : String, action : String);case class NegotiateNewContract;case class ThatsAWrap;def main(args : Array[String]) ={def ct ="Thread " + Thread.currentThread().getName() + ": "val badActor =actor{var done = falsewhile (! done){receive{case NegotiateNewContract =>System.out.println(ct + "I won't do it for less than $1 million!")case Speak(line) =>System.out.println(ct + line)case Gesture(bodyPart, action) =>System.out.println(ct + "(" + action + "s " + bodyPart + ")")case ThatsAWrap =>System.out.println(ct + "Great cast party, everybody! See ya!")done = truecase _ =>System.out.println(ct + "Huh? I'll be in my trailer.")}}}System.out.println(ct + "Negotiating...")badActor ! NegotiateNewContractSystem.out.println(ct + "Speaking...")badActor ! Speak("Do ya feel lucky, punk?")System.out.println(ct + "Gesturing...")badActor ! Gesture("face", "grimaces")System.out.println(ct + "Speaking again...")badActor ! Speak("Well, do ya?")System.out.println(ct + "Wrapping up")badActor ! ThatsAWrap}}

运行这个新示例，就会非常明确地发现确实有两个不同的线程：

main 线程（所有 Java 程序都以它开始）
Thread-2 线程，它是 Scala Actors 库在幕后生成的

因此，在启动第一个 actor 时，本质上已经开始了多线程执行。

但是，习惯这种新的执行模型可能有点儿困难，因为这是一种全新的并发性考虑方式。例如，请考虑前一篇文章中的 Producer/Consumer 模型。那里有大量代码，尤其是在 Drop 类中，我们可以清楚地看到线程之间，以及线程与保证所有东西同步的监视器之间有哪些交互活动。为了便于参考，我在这里给出前一篇文章中的 V3 代码：

清单 5. ProdConSample，v3 (Scala)

package com.tedneward.scalaexamples.scala.V3
{import concurrent.MailBoximport concurrent.ops._object ProdConSample{class Drop{private val m = new MailBox()private case class Empty()private case class Full(x : String)m send Empty()  // initializationdef put(msg : String) : Unit ={m receive{case Empty() =>m send Full(msg)}}def take() : String ={m receive{case Full(msg) =>m send Empty(); msg}}}def main(args : Array[String]) : Unit ={// Create Dropval drop = new Drop()// Spawn Producerspawn{val importantInfo : Array[String] = Array("Mares eat oats","Does eat oats","Little lambs eat ivy","A kid will eat ivy too");importantInfo.foreach((msg) => drop.put(msg))drop.put("DONE")}// Spawn Consumerspawn{var message = drop.take()while (message != "DONE"){System.out.format("MESSAGE RECEIVED: %s%n", message)message = drop.take()}}}}
}

尽管看到 Scala 如何简化这些代码很有意思，但是它实际上与原来的 Java 版本没有概念性差异。现在，看看如果把 Producer/Consumer 示例的基于 actor 的版本缩减到最基本的形式，它会是什么样子：

清单 6. Take 1，开拍！生产！消费！

  object ProdConSample1{case class Message(msg : String)def main(args : Array[String]) : Unit ={val consumer =actor{var done = falsewhile (! done){receive{case msg =>System.out.println("Received message! -> " + msg)done = (msg == "DONE")}}}consumer ! "Mares eat oats"consumer ! "Does eat oats"consumer ! "Little lambs eat ivy"consumer ! "Kids eat ivy too"consumer ! "DONE"      }}

第一个版本确实简短多了，而且在某些情况下可能能够完成所需的所有工作；但是，如果运行这段代码并与以前的版本做比较，就会发现一个重要的差异 — 基于 actor 的版本是一个多位置缓冲区，而不是我们以前使用的单位置缓冲。这看起来是一项改进，而不是缺陷，但是我们要通过对比确认这一点。我们来创建 Drop 的基于 actor 的版本，在这个版本中所有对 put() 的调用必须由对 take() 的调用进行平衡。

幸运的是，Scala Actors 库很容易模拟这种功能。希望让 Producer 一直阻塞，直到 Consumer 接收了消息；实现的方法很简单：让 Producer 一直阻塞，直到它从 Consumer 收到已经接收消息的确认。从某种意义上说，这就是以前的基于监视器的代码所做的，那个版本通过对锁对象使用监视器发送这种信号。

在 Scala Actors 库中，最容易的实现方法是使用 !? 方法而不是 ! 方法（这样就会一直阻塞到收到确认时）。（在 Scala Actors 实现中，每个 Java 线程都是一个 actor，所以回复会发送到与 main 线程隐式关联的邮箱）。这意味着 Consumer 需要发送某种确认；这要使用隐式继承的 reply（它还继承 receive 方法），见清单 7：

清单 7. Take 2，开拍！

  object ProdConSample2{case class Message(msg : String)def main(args : Array[String]) : Unit ={val consumer =actor{var done = falsewhile (! done){receive{case msg =>System.out.println("Received message! -> " + msg)done = (msg == "DONE")reply("RECEIVED")}}}System.out.println("Sending....")consumer !? "Mares eat oats"System.out.println("Sending....")consumer !? "Does eat oats"System.out.println("Sending....")consumer !? "Little lambs eat ivy"System.out.println("Sending....")consumer !? "Kids eat ivy too"System.out.println("Sending....")consumer !? "DONE"      }}

如果喜欢使用 spawn 把 Producer 放在 main() 之外的另一个线程中（这非常接近最初的代码），那么代码可能像清单 8 这样：

清单 8. Take 4，开拍！

  object ProdConSampleUsingSpawn{import concurrent.ops._def main(args : Array[String]) : Unit ={// Spawn Consumerval consumer =actor{var done = falsewhile (! done){receive{case msg =>System.out.println("MESSAGE RECEIVED: " + msg)done = (msg == "DONE")reply("RECEIVED")}}}// Spawn Producerspawn{val importantInfo : Array[String] = Array("Mares eat oats","Does eat oats","Little lambs eat ivy","A kid will eat ivy too","DONE");importantInfo.foreach((msg) => consumer !? msg)}}}

无论从哪个角度来看，基于 actor 的版本都比原来的版本简单多了。读者只要让 actor 和隐含的邮箱自己发挥作用即可。

但是，这并不简单。actor 模型完全颠覆了考虑并发性和线程安全的整个过程；在以前的模型中，我们主要关注共享的数据结构（数据并发性），而现在主要关注操作数据的代码本身的结构（任务并发性），尽可能少共享数据。请注意 Producer/Consumer 示例的不同版本的差异。在以前的示例中，并发功能是围绕 Drop 类（有界限的缓冲区）显式编写的。在本文中的版本中，Drop 甚至没有出现，重点在于两个 actor（线程）以及它们之间的交互（通过不共享任何东西的消息）。

当然，仍然可以用 actor 构建以数据为中心的并发构造；只是必须采用稍有差异的方式。请考虑一个简单的 “计数器” 对象，它使用 actor 消息传达 “increment” 和 “get” 操作，见清单 9：

清单 9. Take 5，计数！

 object CountingSample{case class Incrcase class Value(sender : Actor)case class Lock(sender : Actor)case class UnLock(value : Int)class Counter extends Actor{override def act(): Unit = loop(0)def loop(value: int): Unit = {receive {case Incr()   => loop(value + 1)case Value(a) => a ! value; loop(value)case Lock(a)  => a ! valuereceive { case UnLock(v) => loop(v) }case _        => loop(value)}}}def main(args : Array[String]) : Unit ={val counter = new Countercounter.start()counter ! Incr()counter ! Incr()counter ! Incr()counter ! Value(self)receive { case cvalue => Console.println(cvalue) }    counter ! Incr()counter ! Incr()counter ! Value(self)receive { case cvalue => Console.println(cvalue) }    }}

为了进一步扩展 Producer/Consumer 示例，清单 10 给出一个在内部使用 actor 的 Drop 版本（这样，其他 Java 类就可以使用这个 Drop，而不需要直接调用 actor 的方法）：

清单 10. 在内部使用 actor 的 Drop

  object ActorDropSample{class Drop{private case class Put(x: String)private case object Takeprivate case object Stopprivate val buffer =actor{var data = ""loop{react{case Put(x) if data == "" =>data = x; reply()case Take if data != "" =>val r = data; data = ""; reply(r)case Stop =>reply(); exit("stopped")}}}def put(x: String) { buffer !? Put(x) }def take() : String = (buffer !? Take).asInstanceOf[String]def stop() { buffer !? Stop }}def main(args : Array[String]) : Unit ={import concurrent.ops._// Create Dropval drop = new Drop()// Spawn Producerspawn{val importantInfo : Array[String] = Array("Mares eat oats","Does eat oats","Little lambs eat ivy","A kid will eat ivy too");importantInfo.foreach((msg) => { drop.put(msg) })drop.put("DONE")}// Spawn Consumerspawn{var message = drop.take()while (message != "DONE"){System.out.format("MESSAGE RECEIVED: %s%n", message)message = drop.take()}drop.stop()}}}

可以看到，这需要更多代码（和更多的线程，因为每个 actor 都在一个线程池内部起作用），但是这个版本的 API 与以前的版本相同，它把所有与并发性相关的代码都放在 Drop 内部，这正是 Java 开发人员所期望的。

actor 还有更多特性。

在规模很大的系统中，让每个 actor 都由一个 Java 线程支持是非常浪费资源的，尤其是在 actor 的等待时间比处理时间长的情况下。在这些情况下，基于事件的 actor 可能更合适；这种 actor 实际上放在一个闭包中，闭包捕捉 actor 的其他动作。也就是说，现在并不通过线程状态和寄存器表示代码块（函数）。当一个消息到达 actor 时（这时显然需要活动的线程），触发闭包，闭包在它的活动期间借用一个活动的线程，然后通过回调本身终止或进入 “等待” 状态，这样就会释放线程。（请参见参考资料中 Haller/Odersky 的文章）。

在 Scala Actors 库中，这要使用 react 方法而不是前面使用的 receive。使用 react 的关键是在形式上 react 不能返回，所以 react 中的实现必须重复调用包含 react 块的代码块。简便方法是使用 loop 结构创建一个接近无限的循环。这意味着清单 10 中的 Drop 实现实际上只通过借用调用者的线程执行操作，这会减少执行所有操作所需的线程数。（在实践中，我还没有见过在简单的示例中出现这种效果，所以我想我们只能暂且相信 Scala 设计者的说法）。

在某些情况下，可能选择通过派生基本的 Actor 类（在这种情况下，必须定义 act 方法，否则类仍然是抽象的）创建一个新类，它隐式地作为 actor 执行。尽管这是可行的，但是这种思想在 Scala 社区中不受欢迎；在一般情况下，我在这里描述的方法（使用 Actor 对象中的 actor 方法）是创建 actor 的首选方法。

结束语

因为 actor 编程需要与 “传统” 对象编程不同的风格，所以在使用 actor 时要记住几点。

首先，actor 的主要能力来源于消息传递风格，而不采用阻塞-调用风格，这是它的主要特点。（有意思的是，也有使用消息传递作为核心机制的面向对象语言。最知名的两个例子是 Objective-C 和 Smalltalk，还有 ThoughtWorker 的 Ola Bini 新创建的 Ioke）。如果创建直接或间接扩展 Actor 的类，那么要确保对对象的所有调用都通过消息传递进行。

第二，因为可以在任何时候交付消息，而且更重要的是，在发送和接收之间可能有相当长的延迟，所以一定要确保消息携带正确地处理它们所需的所有状态。这种方式会：

让代码更容易理解（因为消息携带处理所需的所有状态）。
减少 actor 访问某些地方的共享状态的可能性，从而减少发生死锁或其他并发性问题的机会。

第三，actor 应该不会阻塞，您从前面的内容应该能够看出这一点。从本质上说，阻塞是导致死锁的原因；代码可能产生的阻塞越少，发生死锁的可能性就越低。

很有意思的是，如果您熟悉 Java Message Service (JMS) API，就会发现我给出的这些建议在很大程度上也适用于 JMS — 毕竟，actor 消息传递风格只是在实体之间传递消息，JMS 消息传递也是在实体之间传递消息。它们的差异在于，JMS 消息往往比较大，在层和进程级别上操作；而 actor 消息往往比较小，在对象和线程级别上操作。如果您掌握了 JMS，actor 也不难掌握。

actor 并不是解决所有并发性问题的万灵药，但是它们为应用程序或库代码的建模提供了一种新的方式，所用的构造相当简单明了。尽管它们的工作方式有时与您预期的不一样，但是一些行为正是我们所熟悉的 — 毕竟，我们在最初使用对象时也有点不习惯，只要经过努力，您也会掌握并喜欢上 actor。

本期就到这里了；下一期再见！