從Java走進(jìn)Scala:深入了解Scala并發(fā)性

作者：Ted Neward 2009-07-15 10:14:25

對于許多（如果不是大多數(shù)）Java 程序員來說，Scala 的吸引力在于處理并發(fā)性以及編寫線程安全的代碼時非常輕松。在本期文章中，Ted Neward 將開始深入研究 Scala 語言及環(huán)境所提供的各種并發(fā)特性和庫。

2003 年，Herb Sutter 在他的文章 “The Free Lunch Is Over” 中揭露了行業(yè)中最不可告人的一個小秘密，他明確論證了處理器在速度上的發(fā)展已經(jīng)走到了盡頭，并且將由全新的單芯片上的并行 “內(nèi)核”（虛擬 CPU）所取代。這一發(fā)現(xiàn)對編程社區(qū)造成了不小的沖擊，因?yàn)檎_創(chuàng)建線程安全的代碼，在理論而非實(shí)踐中，始終會提高高性能開發(fā)人員的身價，而讓各公司難以聘用他們。看上去，僅有少數(shù)人充分理解了 Java 的線程模型、并發(fā) API 以及 “同步” 的含義，以便能夠編寫同時提供安全性和吞吐量的代碼 —— 并且大多數(shù)人已經(jīng)明白了它的困難所在。

51CTO編輯推薦：Scala編程語言專題

據(jù)推測，行業(yè)的其余部分將自力更生，這顯然不是一個理想的結(jié)局，至少不是 IT 部門努力開發(fā)軟件所應(yīng)得的回報。

與 Scala 在 .NET 領(lǐng)域中的姐妹語言 F# 相似，Scala 是針對 “并發(fā)性問題” 的解決方案之一。在本期文章中，我討論了 Scala 的一些屬性，這些屬性使它更加勝任于編寫線程安全的代碼，比如默認(rèn)不可修改的對象，并討論了一種返回對象副本而不是修改它們內(nèi)容的***設(shè)計方案。Scala 對并發(fā)性的支持遠(yuǎn)比此深遠(yuǎn)；現(xiàn)在，我們有必要來了解一下 Scala 的各種庫。

并發(fā)性基礎(chǔ)

在深入研究 Scala 的并發(fā)性支持之前，有必要確保您具備了對 Java 基本并發(fā)性模型的良好理解，因?yàn)?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 中的代碼進(jìn)行比較，尋找它們之間有哪些不同；他們會發(fā)現(xiàn)我并未 “同步” put 和 take 方法，而是使用了存儲在 Drop 中的 lock 對象。其原因非常簡單：對象的監(jiān)測程序永遠(yuǎ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)封裝了線程安全的實(shí)現(xiàn)；然后，它才能依賴客戶機(jī)的優(yōu)勢正常運(yùn)行。

注意：我在此處展示的代碼對 Sun 教程解決方案做了少許修改；它們提供的代碼存在一個很小的設(shè)計缺陷（參見 Java 教程 “缺陷”）。

Producer/Consumer 問題的核心非常容易理解：一個（或多個）生產(chǎn)者實(shí)體希望將數(shù)據(jù)提供給一個（或多個）使用者實(shí)體供它們使用和操作（在本例中，它包括將數(shù)據(jù)打印到控制臺）。Producer 和 Consumer 類是相應(yīng)直觀的 Runnable-實(shí)現(xiàn)類：Producer 從數(shù)組中獲取 String，并通過 put 將它們放置到 Consumer 的緩沖區(qū)中，并根據(jù)需要執(zhí)行 take。

問題的難點(diǎn)在于，如果 Producer 運(yùn)行過快，則數(shù)據(jù)在覆蓋時可能會丟失；如果 Consumer 運(yùn)行過快，則當(dāng) Consumer 讀取相同的數(shù)據(jù)兩次時，數(shù)據(jù)可能會得到重復(fù)處理。緩沖區(qū)（在 Java Tutorial 代碼中稱作 Drop）將確保不會出現(xiàn)這兩種情況。數(shù)據(jù)破壞的可能性就更不用提了（在 String 引用的例子中很困難，但仍然值得注意），因?yàn)閿?shù)據(jù)會由 put 放入緩沖區(qū)，并由 take 取出。

關(guān)于此主題的全面討論請閱讀 Brian Goetz 的 Java Concurrency in Practice 或 Doug Lea 的 Concurrent Programming in Java（參見參考資料），但是，在應(yīng)用 Scala 之前有必要快速了解一下此代碼的運(yùn)行原理。

當(dāng) 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() 方法將通知當(dāng)前線程進(jìn)入非活動狀態(tài)，并等待另一個線對該對象調(diào)用 notifyAll()。然后，通知的線程必須在能夠繼續(xù)執(zhí)行的時候嘗試再次獲取監(jiān)控程序。從本質(zhì)上說，wait() 和 notify()/notifyAll() 允許一種簡單的信令機(jī)制，它允許 Drop 在 Producer 和 Consumer 線程之間進(jìn)行協(xié)調(diào)，每個 put 都有相應(yīng)的 take。

本文的代碼下載部分使用 Java5 并發(fā)性增強(qiáng)（Lock 和 Condition 接口以及 ReentrantLock 鎖定實(shí)現(xiàn)）提供清單 2 的基于超時的版本，但基本代碼模式仍然相同。這就是問題所在：編寫清單 2 這樣的代碼的開發(fā)人員需要過度專注于線程和鎖定的細(xì)節(jié)以及低級實(shí)現(xiàn)代碼，以便讓它們能夠正確運(yùn)行。此外，開發(fā)人員需要對每一行代碼刨根知底，以確定是否需要保護(hù)它們，因?yàn)檫^度同步與過少同步同樣有害。

現(xiàn)在，我們來看到 Scala 替代方案。

#p#

良好的 Scala 并發(fā)性 (v1)

開始應(yīng)用 Scala 并發(fā)性的一種方法是將 Java 代碼直接轉(zhuǎn)換為 Scala，以便利用 Scala 的語法優(yōu)勢來簡化代碼（至少能簡化一點(diǎn)）：

清單 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 同類相同，再一次擴(kuò)展（實(shí)現(xiàn)）了 Runnable 接口并覆蓋了 run() 方法，并且 — 對于 Producer 的情況 — 分別使用了內(nèi)置迭代方法來遍歷 importantInfo 數(shù)組的內(nèi)容。（實(shí)際上，為了讓它更像 Scala，importantInfo 可能應(yīng)該是一個 List 而不是 Array，但在***次嘗試時，我希望盡可能保證它們與原始 Java 代碼一致。）

Drop 類同樣類似于它的 Java 版本。但 Scala 中有一些例外，“synchronized” 并不是關(guān)鍵字，它是針對 AnyRef 類定義的一個方法，即 Scala “所有引用類型的根”。這意味著，要同步某個特定的對象，您只需要對該對象調(diào)用同步方法；在本例中，對 Drop 上的 lock 字段中所保存的對象調(diào)用同步方法。

注意，我們在 await() 方法定義的 Drop 類中還利用了一種 Scala 機(jī)制：cond 參數(shù)是等待計算的代碼塊，而不是在傳遞給該方法之前進(jìn)行計算。在 Scala 中，這被稱作 “call-by-name”；此處，它是一種實(shí)用的方法，可以捕獲需要在 Java 版本中表示兩次的條件等待邏輯（分別用于 put 和 take）。

***，在 main() 中，創(chuàng)建 Drop 實(shí)例，實(shí)例化兩個線程，使用 start() 啟動它們，然后在 main() 的結(jié)束部分退出，相信 JVM 會在 main() 結(jié)束之前啟動這兩個線程。（在生產(chǎn)代碼中，可能無法保證這種情況，但對于這樣的簡單的例子，99.99％沒有問題。）

但是，已經(jīng)說過，仍然存在相同的基本問題：程序員仍然需要過分擔(dān)心兩個線程之間的通信和協(xié)調(diào)問題。雖然一些 Scala 機(jī)制可以簡化語法，但這目前為止并沒有相當(dāng)大的吸引力。

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ù)的細(xì)節(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 的實(shí)現(xiàn)，它現(xiàn)在利用 MailBox 類處理傳入以及從 Drop 中刪除的消息的阻塞和信號事務(wù)。（我們可以重寫 Producer 和 Consumer，讓它們直接使用 MailBox，但考慮到簡單性，我們假定希望保持所有示例中的 Drop API 相一致。）使用 MailBox 與使用典型的 BoundedBuffer（Drop）稍有不同，因此我們來仔細(xì)看看其代碼。

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 實(shí)例，因此會阻塞直到發(fā)送 Empty。此時，它發(fā)送一個 Full 實(shí)例給包含新數(shù)據(jù)的 MailBox。

take 方法，由于它會從 Drop 中刪除數(shù)據(jù)，對 MailBox 調(diào)用 receive() 以查找 Full 實(shí)例，提取消息（再次得益于模式匹配從 case 類內(nèi)部提取值并將它們綁到本地變量的能力）并發(fā)送一個 Empty 實(shí)例給 MailBox。

不需要明確的鎖定，并且不需要考慮監(jiān)控程序。

#p#

Scala 并發(fā)性 v3

事實(shí)上，我們可以顯著縮短代碼，只要 Producer 和 Consumer 不需要功能全面的類（此處便是如此） — 兩者從本質(zhì)上說都是 Runnable.run() 方法的瘦包裝器，Scala 可以使用 scala.concurrent.ops 對象的 spawn 方法來實(shí)現(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 對象導(dǎo)入）接收一個代碼塊（另一個 by-name 參數(shù)示例）并將它包裝在匿名構(gòu)造的線程對象的 run() 方法內(nèi)部。事實(shí)上，并不難理解 spawn 的定義在 ops 類的內(nèi)部是什么樣的：

清單 6. scala.concurrent.ops.spawn()

def spawn(p: => Unit) = {  
  val t = new Thread() { override def run() = p }  
  t.start()  
}

……這再一次強(qiáng)調(diào)了 by-name 參數(shù)的強(qiáng)大之處。

ops.spawn 方法的一個缺點(diǎn)在于，它是在 2003 年 Java 5 concurrency 類還不可用的時候編寫的。特別是，java.util.concurrent.Executor 及其同類的作用是讓開發(fā)人員更加輕松地生成線程，而不需要實(shí)際處理直接創(chuàng)建線程對象的細(xì)節(jié)。幸運(yùn)的是，在您自己的自定義庫中重新創(chuàng)建 spawn 的定義是相當(dāng)簡單的，這需要利用 Executor（或 ExecutorService 或 ScheduledExecutorService）來執(zhí)行線程的實(shí)際啟動任務(wù)。

事實(shí)上，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()）。

其次，這些基本機(jī)制上面有一個抽象層，詳見本文所討論的 MailBox 類以及將在本系列下一篇文章中討論的 Actors 庫。

兩個例子中的目標(biāo)是相同的：讓開發(fā)人員能夠更加輕松地專注于問題的實(shí)質(zhì)，而不用考慮并發(fā)編程的低級細(xì)節(jié)（顯然，第二種方法更好地實(shí)現(xiàn)了這一目標(biāo)，至少對于沒有過多考慮低級細(xì)節(jié)的人來說是這樣的。）

但是，當(dāng)前 Scala 庫的一個明顯的缺陷就是缺乏 Java 5 支持；scala.concurrent.ops 類應(yīng)該具有 spawn 這樣的利用新的 Executor 接口的方法。它還應(yīng)該支持利用新的 Lock 接口的各種版本的 synchronized。幸運(yùn)的是，這些都是可以在 Scala 生命周期中實(shí)現(xiàn)的庫增強(qiáng)，而不會破壞已有代碼；它們甚至可以由 Scala 開發(fā)人員自己完成，而不需要等待 Scala 的核心開發(fā)團(tuán)隊(duì)提供給他們（只需要花費(fèi)少量時間）。

【相關(guān)閱讀】