7.actor

Scala中處理併發，有很多選擇：

lactor訊息模型，類似Erlang，首選，Lift和akka也實現了自己的actor模型。

lThread、Runnable

ljava.util.concurennt

l3^rd併發框架如Netty，Mina

7.1.actor模型

Java內建執行緒模型	Scala actor模型
“共享資料-鎖”模型（share data and lock）	share nothing
每個object有一個monitor，監視多執行緒對共享資料的訪問	不共享資料，actor之間通過message通訊
加鎖的程式碼段用 synchronized標識
死鎖問題
每個執行緒內部是順序執行的	每個actor內部是順序執行的

7.2.多核計算

對比如下的演算法：

def perfect(n:Int) = n==(1 until n filter (n%_==0) sum) val n = 33550336 // 序列計算 n to n+10 foreach (i=>println(perfect(i)))	def perfect(n:Int) = n==(1 until n filter (n%_==0) sum) val n = 33550336 // 平行計算 class T1(n:Int) extends Thread {   override def run(){println(perfect(n))}} n to n+10 foreach (i=>new T1(i).start)
耗時：8297	耗時：5134
單執行緒序列計算，不能很好發揮多核優勢	多執行緒平行計算，平均分配到多核，更快

上面是Java的寫法，也可以用Scala的actor寫法：

Scala寫法1：

import actors.Actor,actors.Actor._

class A1 extends Actor {

def act { react { case n:Int=>println(perfect(n)) }}}

n to n+10 foreach (i=>{ (new A1).start ! i})

Scala寫法2：

val aa = Array.fill(11)(actor { react { case n:Int=>println(perfect(n)) }})

n to n+10 foreach (i=>aa(i-n) ! i)

或者：

n to n+10 foreach (i=> actor { react { case n:Int=>println(perfect(n)) }} ! i)

7.3.Actor用法

Scala會建立一個執行緒池共所有Actor來使用。receive模型是Actor從池中取一個執行緒一直使用；react模型是Actor從池中取一個執行緒用完給其他Actor用

實現方式1：

importscala.actors._

objectActor1extendsActor { // 或者class

// 實現執行緒

defact() { react { case _ =>println("ok"); exit} }

}

//傳送訊息：

Actor1.start ! 1001 // 必須呼叫start

實現方式2：

importscala.actors.Actor._

vala2 = actor { react { case _ =>println("ok") } } // 馬上啟動

//傳送訊息：

a2 ! "message" // 不必呼叫start

提示：

!	傳送非同步訊息，沒有返回值。
!?	傳送同步訊息，等待返回值。（會阻塞傳送訊息語句所在的執行緒）
!!	傳送非同步訊息，返回值是 Future[Any]。
?	不帶引數。檢視 mailbox 中的下一條訊息。

7.4.方式1：接受receive

特點：要反覆處理訊息，receive外層用while(..)

import actors.Actor, actors.Actor._

val a1 = Actor.actor {

varwork = true

while(work) {

receive { // 接受訊息, 或者用receiveWith(1000)

casemsg:String => println("a1: "+msg)

casex:Int => work = false;println("a1 stop: "+x)

}

}

}

a1 ! "hello" // "a1: hello"

a1 ! "world" // "a1: world"

a1 ! -1 // "a1 stop: -1"

a1 ! "no response :("

7.5.方式2：接受react, loop

特點：

l從不返回

l要反覆執行訊息處理，react外層用loop，不能用while(..);

l通過複用執行緒，比receive更高效，應儘可能使用react

import actors.Actor, Actor._

val a1 = Actor.actor {

react {

case x:Int => println("a1 stop: " + x)

case msg:String => println("a1: " + msg); act()

    }

}

a1 ! "hello" // "a1: hello"

a1 ! "world" // "a1: world"

a1 ! -1 // "a1 stop: -1"

a1 ! "no response :("

如果不用退出的執行緒，可使用loop改寫如下：

val a1 = Actor.actor {

  loop {

react {

casex:Int => println("a1 stop: "+x); exit()

case msg:String => println("a1: " + msg)

    }

  }

}

7.6.REPL接受訊息

scala> self ! "hello"

scala> self.receive { case x => x }

scala> self.receiveWithin(1000) { case x => x }

7.7.actor最佳實踐

7.7.1.不阻塞actor

actor不應由於處理某條訊息而阻塞，可以呼叫helper-actor處理耗時操作(helper actor雖然是阻塞的，但由於不接受訊息所以沒問題)，以便actor接著處理下一條訊息

-----------------------------------------

import actors._, actors.Actor._

val time = 1000

// （1）原來阻塞的程式

val mainActor1 = actor {

    loop { react {

case n: Int => Thread.sleep(time)

                         println(n)

case s => println(s) } }

  }

  1 to 5 foreach { mainActor1 ! _ } // 5秒鐘後列印完數字

// （2）改寫由helper actor去阻塞的程式

val mainActor2: Actor = actor {

    loop { react {

case n: Int => actor { Thread.sleep(time); mainActor2 ! "wakeup" }

                        println(n)

case s => println(s) } }

  }

  1 to 5 foreach { mainActor2 ! _ } // 馬上列印數字; 1秒鐘後列印5個wakeup

-----------------------------------------

7.7.2.actor之間用且僅用訊息來通訊

actor模型讓我們寫多執行緒程式時只用關注各個獨立的單執行緒程式（actor），他們之間通過訊息來通訊。例如，如果BadActor中有一個GoodActor的引用:

class BadActor(a:GoodActor) extends Actor {...}

那在BadActor中即可以通過該引用來直接呼叫GoodActor的方法，也可以通過“!”來傳遞訊息。選擇後者！因為一旦BadActor通過引用讀取GoodActor例項的私有資料，而這些資料可能正被其他執行緒改寫值，結果就避免不了“共享資料-鎖”模型中的麻煩事：即必須保證BadActor執行緒讀取GoodActor的私有資料時，GoodActor執行緒在這塊成為“共享資料”的操作上加鎖。GoodActor只要有了共享資料，就必須來加鎖防範競用衝突和死鎖，你又得從actor模型退回到“共享資料-鎖”模型（注：actor對訊息是順序處理的，本來不用考慮共享資料）。

7.7.3.採用不可變訊息

Scala的actor模型讓每個actor的act方法內部接近於單執行緒環境，你不用當心act方法裡面的操作是否執行緒安全。在act方法中你可以盡情使用非同步、可變物件，因為每個act方法被有效限制在單個執行緒中，這也是actor模型被稱為“share-nothing”模型（零共享模型）的原因，其資料的作用範圍被限制在單個執行緒中。不過一旦物件內的資料被用於多個actor之間進行訊息傳遞。這時你就必須考慮訊息物件是否執行緒安全。

保證訊息物件執行緒安全的最好方法就是保證只使用不可變物件作為訊息物件。訊息類中只定義val欄位，且只能指向不可變物件。定義這種不可變訊息類的簡單方法就是使用case class，並保證其所有的val欄位都是不可變的。Scala API中提供了很多不可變物件可用，例如基本型別、String、Tuple、List，不可變Set、不可變Map等。

如果你發現確實需要把一個可變物件obj1傳送給其他actor，也因該是傳送一份拷貝物件obj1.clone過去，而不是把obj1直接發過去。例如，資料物件Array是可變且未做同步的，所以Array只應該由一個actor同時存取，如果需要傳送陣列arr，就傳送arr.clone（arr中的元素也應該是不可變物件），或者直接傳送一個不可變物件arr.toList更好。

總結：大部分時候使用不可變物件很方便，不可變物件是並行系統的曙光，它們是易使用、低風險的執行緒安全物件。當你將來要設計一個和並行相關的程式時，無論是否使用actor，都應該儘量使用不可變的資料結構。

7.7.4.讓訊息自說明

對每一種訊息建立一個對應的case class，而不是使用上面的tuple資料結構。雖然這種包裝在很多情況下並非必須，但該做法能使actor程式易於理解，例如：

// 不易理解，因為傳遞的是個一般的字串，很難指出那個actor來響應這個訊息

lookerUpper ! ("www.scala-lang.org", self)

// 改為如下，則指出只有react能處理LoopupIP的actor來處理：

caseclass LookupIP(hostname: String, requester: Actor)

lookerUpper ! LookupIP("www.scala-lang.org", self)

7.8.不同jvm間的訊息訪問

伺服器端：

objectActorServerextendsApplication {

importactors.Actor, actors.Actor._, actors.remote.RemoteActor

Actor.actor { // 建立並啟動一個 actor

// 當前 actor 監聽的埠： 3000

RemoteActor.alive(3000)

// 在 3000 埠註冊本 actor，取名為 server1。

// 第一個引數為 actor 的標識，它以單引號開頭，是 Scala 中的 Symbol 量，

// Symbol 量和字串相似，但 Symbol 相等是基於字串比較的。

// self 指代當前 actor （注意此處不能用 this）

RemoteActor.register('server1, Actor.self)

// 收到訊息後的響應

loop {

Actor.react {casemsg =>

println("server1 get: "+msg)

        }

相關推薦