從Java走進Scala:深入了解Scala并發(fā)性
2003 年，Herb Sutter 在他的文章 “The Free Lunch Is Over” 中揭露了行業(yè)中最不可告人的一個小秘密，他明確論證了處理器在速度上的發(fā)展已經(jīng)走到了盡頭，并且將由全新的單芯片上的并行 “內(nèi)核”（虛擬 CPU）所取代。這一發(fā)現(xiàn)對編程社區(qū)造成了不小的沖擊，因為正確創(chuàng)建線程安全的代碼，在理論而非實踐中，始終會提高高性能開發(fā)人員的身價，而讓各公司難以聘用他們?？瓷先ィ瑑H有少數(shù)人充分理解了 Java 的線程模型、并發(fā) API 以及 “同步” 的含義，以便能夠編寫同時提供安全性和吞吐量的代碼 —— 并且大多數(shù)人已經(jīng)明白了它的困難所在。
51CTO編輯推薦：Scala編程語言專題
據(jù)推測，行業(yè)的其余部分將自力更生，這顯然不是一個理想的結(jié)局，至少不是 IT 部門努力開發(fā)軟件所應(yīng)得的回報。
與 Scala 在 .NET 領(lǐng)域中的姐妹語言 F# 相似，Scala 是針對 “并發(fā)性問題” 的解決方案之一。在本期文章中，我討論了 Scala 的一些屬性，這些屬性使它更加勝任于編寫線程安全的代碼，比如默認不可修改的對象，并討論了一種返回對象副本而不是修改它們內(nèi)容的***設(shè)計方案。Scala 對并發(fā)性的支持遠比此深遠；現(xiàn)在，我們有必要來了解一下 Scala 的各種庫。
并發(fā)性基礎(chǔ)
在深入研究 Scala 的并發(fā)性支持之前，有必要確保您具備了對 Java 基本并發(fā)性模型的良好理解，因為 Scala 的并發(fā)性支持，從某種程度上說，建立在 JVM 和支持庫所提供的特性和功能的基礎(chǔ)之上。為此，清單 1 中的代碼包含了一個已知的 Producer/Consumer 并發(fā)性問題（詳見 Sun Java Tutorial 的 “Guarded Blocks” 小節(jié)）。注意，Java Tutorial 版本并未在其解決方案中使用 java.util.concurrent 類，而是擇優(yōu)使用了 java.lang.Object 中的較舊的 wait()/notifyAll() 方法：
清單 1. Producer/Consumer（Java5 之前）
- package com.tedneward.scalaexamples.notj5;
- class Producer implements Runnable
- {
- private Drop drop;
- private String importantInfo[] = {
- "Mares eat oats",
- "Does eat oats",
- "Little lambs eat ivy",
- "A kid will eat ivy too"
- };
- public Producer(Drop drop) { this.drop = drop; }
- public void run()
- {
- for (int i = 0; i < importantInfo.length; i++)
- {
- drop.put(importantInfo[i]);
- }
- drop.put("DONE");
- }
- }
- class Consumer implements Runnable
- {
- private Drop drop;
- public Consumer(Drop drop) { this.drop = drop; }
- public void run()
- {
- for (String message = drop.take(); !message.equals("DONE");
- message = drop.take())
- {
- System.out.format("MESSAGE RECEIVED: %s%n", message);
- }
- }
- }
- class Drop
- {
- //Message sent from producer to consumer.
- private String message;
- //True if consumer should wait for producer to send message,
- //false if producer should wait for consumer to retrieve message.
- private boolean empty = true;
- //Object to use to synchronize against so as to not "leak" the
- //"this" monitor
- private Object lock = new Object();
- public String take()
- {
- synchronized(lock)
- {
- //Wait until message is available.
- while (empty)
- {
- try
- {
- lock.wait();
- }
- catch (InterruptedException e) {}
- }
- //Toggle status.
- empty = true;
- //Notify producer that status has changed.
- lock.notifyAll();
- return message;
- }
- }
- public void put(String message)
- {
- synchronized(lock)
- {
- //Wait until message has been retrieved.
- while (!empty)
- {
- try
- {
- lock.wait();
- } catch (InterruptedException e) {}
- }
- //Toggle status.
- empty = false;
- //Store message.
- this.message = message;
- //Notify consumer that status has changed.
- lock.notifyAll();
- }
- }
- }
- public class ProdConSample
- {
- public static void main(String[] args)
- {
- Drop drop = new Drop();
- (new Thread(new Producer(drop))).start();
- (new Thread(new Consumer(drop))).start();
- }
- }
Java 教程 “缺陷”
好奇的讀者可能會將此處的代碼與 Java Tutorial 中的代碼進行比較，尋找它們之間有哪些不同；他們會發(fā)現(xiàn)我并未 “同步” put 和 take 方法，而是使用了存儲在 Drop 中的 lock 對象。其原因非常簡單：對象的監(jiān)測程序永遠都不會封裝在類的內(nèi)部，因此 Java Tutorial 版本允許此代碼打破此規(guī)則（顯然很瘋狂）：
- public class ProdConSample
- {
- public static void main(String[] args)
- {
- Drop drop = new Drop();
- (new Thread(new Producer(drop))).start();
- (new Thread(new Consumer(drop))).start();
- synchronized(drop)
- {
- Thread.sleep(60 * 60 * 24 * 365 * 10); // sleep for 10 years?!?
- }
- }
- }
通過使用私有對象作為鎖定所依托的監(jiān)測程序，此代碼將不會有任何效果。從本質(zhì)上說，現(xiàn)在已經(jīng)封裝了線程安全的實現(xiàn)；然后，它才能依賴客戶機的優(yōu)勢正常運行。
注意：我在此處展示的代碼對 Sun 教程解決方案做了少許修改；它們提供的代碼存在一個很小的設(shè)計缺陷（參見 Java 教程 “缺陷”）。
Producer/Consumer 問題的核心非常容易理解：一個（或多個）生產(chǎn)者實體希望將數(shù)據(jù)提供給一個（或多個）使用者實體供它們使用和操作（在本例中，它包括將數(shù)據(jù)打印到控制臺）。Producer 和 Consumer 類是相應(yīng)直觀的 Runnable-實現(xiàn)類：Producer 從數(shù)組中獲取 String，并通過 put 將它們放置到 Consumer 的緩沖區(qū)中，并根據(jù)需要執(zhí)行 take。
問題的難點在于，如果 Producer 運行過快，則數(shù)據(jù)在覆蓋時可能會丟失；如果 Consumer 運行過快，則當 Consumer 讀取相同的數(shù)據(jù)兩次時，數(shù)據(jù)可能會得到重復(fù)處理。緩沖區(qū)（在 Java Tutorial 代碼中稱作 Drop）將確保不會出現(xiàn)這兩種情況。數(shù)據(jù)破壞的可能性就更不用提了（在 String 引用的例子中很困難，但仍然值得注意），因為數(shù)據(jù)會由 put 放入緩沖區(qū)，并由 take 取出。
關(guān)于此主題的全面討論請閱讀 Brian Goetz 的 Java Concurrency in Practice 或 Doug Lea 的 Concurrent Programming in Java（參見 參考資料），但是，在應(yīng)用 Scala 之前有必要快速了解一下此代碼的運行原理。
當 Java 編譯器看到 synchronized 關(guān)鍵字時，它會在同步塊的位置生成一個 try/finally 塊，其頂部包括一個 monitorenter 操作碼，并且 finally 塊中包括一個 monitorexit 操作碼，以確保監(jiān)控程序（Java 的原子性基礎(chǔ)）已經(jīng)發(fā)布，而與代碼退出的方式無關(guān)。因此，Drop 中的 put 代碼將被重寫，如清單 2 所示：
清單 2. 編譯器失效后的 Drop.put
- // This is pseudocode
- public void put(String message)
- {
- try
- {
- monitorenter(lock)
- //Wait until message has been retrieved.
- while (!empty)
- {
- try
- {
- lock.wait();
- } catch (InterruptedException e) {}
- }
- //Toggle status.
- empty = false;
- //Store message.
- this.message = message;
- //Notify consumer that status has changed.
- lock.notifyAll();
- }
- finally
- {
- monitorexit(lock)
- }
- }
wait() 方法將通知當前線程進入非活動狀態(tài)，并等待另一個線對該對象調(diào)用 notifyAll()。然后，通知的線程必須在能夠繼續(xù)執(zhí)行的時候嘗試再次獲取監(jiān)控程序。從本質(zhì)上說，wait() 和 notify()/notifyAll() 允許一種簡單的信令機制，它允許 Drop 在 Producer 和 Consumer 線程之間進行協(xié)調(diào)，每個 put 都有相應(yīng)的 take。
本文的 代碼下載 部分使用 Java5 并發(fā)性增強（Lock 和 Condition 接口以及 ReentrantLock 鎖定實現(xiàn)）提供 清單 2 的基于超時的版本，但基本代碼模式仍然相同。這就是問題所在：編寫清單 2 這樣的代碼的開發(fā)人員需要過度專注于線程和鎖定的細節(jié)以及低級實現(xiàn)代碼，以便讓它們能夠正確運行。此外，開發(fā)人員需要對每一行代碼刨根知底，以確定是否需要保護它們，因為過度同步與過少同步同樣有害。
現(xiàn)在，我們來看到 Scala 替代方案。
#p#
良好的 Scala 并發(fā)性 (v1)
開始應(yīng)用 Scala 并發(fā)性的一種方法是將 Java 代碼直接轉(zhuǎn)換為 Scala，以便利用 Scala 的語法優(yōu)勢來簡化代碼（至少能簡化一點）：
清單 3. ProdConSample (Scala)
- object ProdConSample
- {
- class Producer(drop : Drop)
- extends Runnable
- {
- val importantInfo : Array[String] = Array(
- "Mares eat oats",
- "Does eat oats",
- "Little lambs eat ivy",
- "A kid will eat ivy too"
- );
- override def run() : Unit =
- {
- importantInfo.foreach((msg) => drop.put(msg))
- drop.put("DONE")
- }
- }
- class Consumer(drop : Drop)
- extends Runnable
- {
- override def run() : Unit =
- {
- var message = drop.take()
- while (message != "DONE")
- {
- System.out.format("MESSAGE RECEIVED: %s%n", message)
- message = drop.take()
- }
- }
- }
- class Drop
- {
- var message : String = ""
- var empty : Boolean = true
- var lock : AnyRef = new Object()
- def put(x: String) : Unit =
- lock.synchronized
- {
- // Wait until message has been retrieved
- await (empty == true)
- // Toggle status
- empty = false
- // Store message
- message = x
- // Notify consumer that status has changed
- lock.notifyAll()
- }
- def take() : String =
- lock.synchronized
- {
- // Wait until message is available.
- await (empty == false)
- // Toggle status
- empty=true
- // Notify producer that staus has changed
- lock.notifyAll()
- // Return the message
- message
- }
- private def await(cond: => Boolean) =
- while (!cond) { lock.wait() }
- }
- def main(args : Array[String]) : Unit =
- {
- // Create Drop
- val drop = new Drop();
- // Spawn Producer
- new Thread(new Producer(drop)).start();
- // Spawn Consumer
- new Thread(new Consumer(drop)).start();
- }
- }
Producer 和 Consumer 類幾乎與它們的 Java 同類相同，再一次擴展（實現(xiàn)）了 Runnable 接口并覆蓋了 run() 方法，并且 — 對于 Producer 的情況 — 分別使用了內(nèi)置迭代方法來遍歷 importantInfo 數(shù)組的內(nèi)容。（實際上，為了讓它更像 Scala，importantInfo 可能應(yīng)該是一個 List 而不是 Array，但在***次嘗試時，我希望盡可能保證它們與原始 Java 代碼一致。）
Drop 類同樣類似于它的 Java 版本。但 Scala 中有一些例外，“synchronized” 并不是關(guān)鍵字，它是針對 AnyRef 類定義的一個方法，即 Scala “所有引用類型的根”。這意味著，要同步某個特定的對象，您只需要對該對象調(diào)用同步方法；在本例中，對 Drop 上的 lock 字段中所保存的對象調(diào)用同步方法。
注意，我們在 await() 方法定義的 Drop 類中還利用了一種 Scala 機制：cond 參數(shù)是等待計算的代碼塊，而不是在傳遞給該方法之前進行計算。在 Scala 中，這被稱作 “call-by-name”；此處，它是一種實用的方法，可以捕獲需要在 Java 版本中表示兩次的條件等待邏輯（分別用于 put 和 take）。
***，在 main() 中，創(chuàng)建 Drop 實例，實例化兩個線程，使用 start() 啟動它們，然后在 main() 的結(jié)束部分退出，相信 JVM 會在 main() 結(jié)束之前啟動這兩個線程。（在生產(chǎn)代碼中，可能無法保證這種情況，但對于這樣的簡單的例子，99.99％ 沒有問題。）
但是，已經(jīng)說過，仍然存在相同的基本問題：程序員仍然需要過分擔心兩個線程之間的通信和協(xié)調(diào)問題。雖然一些 Scala 機制可以簡化語法，但這目前為止并沒有相當大的吸引力。
Scala 并發(fā)性 v2
Scala Library Reference 中有一個有趣的包：scala.concurrency。這個包包含許多不同的并發(fā)性結(jié)構(gòu)，包括我們即將利用的 MailBox 類。
顧名思義，MailBox 從本質(zhì)上說就是 Drop，用于在檢測之前保存數(shù)據(jù)塊的單槽緩沖區(qū)。但是，MailBox ***的優(yōu)勢在于它將發(fā)送和接收數(shù)據(jù)的細節(jié)完全封裝到模式匹配和 case 類中，這使它比簡單的 Drop（或 Drop 的多槽數(shù)據(jù)保存類 java.util.concurrent.BoundedBuffer）更加靈活。
清單 4. ProdConSample, v2 (Scala)
- package com.tedneward.scalaexamples.scala.V2
- {
- import concurrent.{MailBox, ops}
- object ProdConSample
- {
- class Producer(drop : Drop)
- extends Runnable
- {
- val importantInfo : Array[String] = Array(
- "Mares eat oats",
- "Does eat oats",
- "Little lambs eat ivy",
- "A kid will eat ivy too"
- );
- override def run() : Unit =
- {
- importantInfo.foreach((msg) => drop.put(msg))
- drop.put("DONE")
- }
- }
- class Consumer(drop : Drop)
- extends Runnable
- {
- override def run() : Unit =
- {
- var message = drop.take()
- while (message != "DONE")
- {
- System.out.format("MESSAGE RECEIVED: %s%n", message)
- message = drop.take()
- }
- }
- }
- class Drop
- {
- private val m = new MailBox()
- private case class Empty()
- private case class Full(x : String)
- m send Empty() // initialization
- def 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 Drop
- val drop = new Drop()
- // Spawn Producer
- new Thread(new Producer(drop)).start();
- // Spawn Consumer
- new Thread(new Consumer(drop)).start();
- }
- }
- }
此處，v2 和 v1 之間的惟一區(qū)別在于 Drop 的實現(xiàn)，它現(xiàn)在利用 MailBox 類處理傳入以及從 Drop 中刪除的消息的阻塞和信號事務(wù)。（我們可以重寫 Producer 和 Consumer，讓它們直接使用 MailBox，但考慮到簡單性，我們假定希望保持所有示例中的 Drop API 相一致。）使用 MailBox 與使用典型的 BoundedBuffer（Drop）稍有不同，因此我們來仔細看看其代碼。
MailBox 有兩個基本操作：send 和 receive。receiveWithin 方法僅僅是基于超時的 receive。MailBox 接收任何類型的消息。send() 方法將消息放置到郵箱中，并立即通知任何關(guān)心該類型消息的等待接收者，并將它附加到一個消息鏈表中以便稍后檢索。receive() 方法將阻塞，直到接收到對于功能塊合適的消息。
因此，在這種情況下，我們將創(chuàng)建兩個 case 類，一個不包含任何內(nèi)容（Empty），這表示 MailBox 為空，另一個包含消息數(shù)據(jù)（Full。
put 方法，由于它會將數(shù)據(jù)放置在 Drop 中，對 MailBox 調(diào)用 receive() 以查找 Empty 實例，因此會阻塞直到發(fā)送 Empty。此時，它發(fā)送一個 Full 實例給包含新數(shù)據(jù)的 MailBox。
take 方法，由于它會從 Drop 中刪除數(shù)據(jù)，對 MailBox 調(diào)用 receive() 以查找 Full 實例，提取消息（再次得益于模式匹配從 case 類內(nèi)部提取值并將它們綁到本地變量的能力）并發(fā)送一個 Empty 實例給 MailBox。
不需要明確的鎖定，并且不需要考慮監(jiān)控程序。
#p#
Scala 并發(fā)性 v3
事實上，我們可以顯著縮短代碼，只要 Producer 和 Consumer 不需要功能全面的類（此處便是如此） — 兩者從本質(zhì)上說都是 Runnable.run() 方法的瘦包裝器，Scala 可以使用 scala.concurrent.ops 對象的 spawn 方法來實現(xiàn)，如清單 5 所示：
清單 5. ProdConSample, v3 (Scala)
- package com.tedneward.scalaexamples.scala.V3
- {
- import concurrent.MailBox
- import concurrent.ops._
- object ProdConSample
- {
- class Drop
- {
- private val m = new MailBox()
- private case class Empty()
- private case class Full(x : String)
- m send Empty() // initialization
- def 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 Drop
- val drop = new Drop()
- // Spawn Producer
- spawn
- {
- 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 Consumer
- spawn
- {
- var message = drop.take()
- while (message != "DONE")
- {
- System.out.format("MESSAGE RECEIVED: %s%n", message)
- message = drop.take()
- }
- }
- }
- }
- }
spawn 方法（通過包塊頂部的 ops 對象導入）接收一個代碼塊（另一個 by-name 參數(shù)示例）并將它包裝在匿名構(gòu)造的線程對象的 run() 方法內(nèi)部。事實上，并不難理解 spawn 的定義在 ops 類的內(nèi)部是什么樣的：
清單 6. scala.concurrent.ops.spawn()
- def spawn(p: => Unit) = {
- val t = new Thread() { override def run() = p }
- t.start()
- }
……這再一次強調(diào)了 by-name 參數(shù)的強大之處。
ops.spawn 方法的一個缺點在于，它是在 2003 年 Java 5 concurrency 類還不可用的時候編寫的。特別是，java.util.concurrent.Executor 及其同類的作用是讓開發(fā)人員更加輕松地生成線程，而不需要實際處理直接創(chuàng)建線程對象的細節(jié)。幸運的是，在您自己的自定義庫中重新創(chuàng)建 spawn 的定義是相當簡單的，這需要利用 Executor（或 ExecutorService 或 ScheduledExecutorService）來執(zhí)行線程的實際啟動任務(wù)。
事實上，Scala 的并發(fā)性支持超越了 MailBox 和 ops 類；Scala 還支持一個類似的 “Actors” 概念，它使用了與 MailBox 所采用的方法相類似的消息傳遞方法，但應(yīng)用更加全面并且靈活性也更好。但是，這部分內(nèi)容將在下期討論。
結(jié)束語
Scala 為并發(fā)性提供了兩種級別的支持，這與其他與 Java 相關(guān)的主題極為類似：
首先，對底層庫的完全訪問（比如說 java.util.concurrent）以及對 “傳統(tǒng)” Java 并發(fā)性語義的支持（比如說監(jiān)控程序和 wait()/notifyAll()）。
其次，這些基本機制上面有一個抽象層，詳見本文所討論的 MailBox 類以及將在本系列下一篇文章中討論的 Actors 庫。
兩個例子中的目標是相同的：讓開發(fā)人員能夠更加輕松地專注于問題的實質(zhì)，而不用考慮并發(fā)編程的低級細節(jié)（顯然，第二種方法更好地實現(xiàn)了這一目標，至少對于沒有過多考慮低級細節(jié)的人來說是這樣的。）
但是，當前 Scala 庫的一個明顯的缺陷就是缺乏 Java 5 支持；scala.concurrent.ops 類應(yīng)該具有 spawn 這樣的利用新的 Executor 接口的方法。它還應(yīng)該支持利用新的 Lock 接口的各種版本的 synchronized。幸運的是，這些都是可以在 Scala 生命周期中實現(xiàn)的庫增強，而不會破壞已有代碼；它們甚至可以由 Scala 開發(fā)人員自己完成，而不需要等待 Scala 的核心開發(fā)團隊提供給他們（只需要花費少量時間）。
【相關(guān)閱讀】
