一、基本介紹

1、 概念

        Scala中的Actor能夠實現並行程式設計的強大功能，它是基於事件模型的併發機制，Scala是運用訊息（message）的傳送、接收來實現多執行緒的。使用Scala能夠更容易地實現多執行緒應用的開發。

2、傳統java併發程式設計與Scala Actor程式設計的區別

        對於Java，我們都知道它的多執行緒實現需要對共享資源（變數、物件等）使用synchronized 關鍵字進行程式碼塊同步、物件鎖互斥等等。而且，常常一大塊的try…catch語句塊中加上wait方法、notify方法、notifyAll方法是讓人很頭疼的。原因就在於Java中多數使用的是可變狀態的物件資源，對這些資源進行共享來實現多執行緒程式設計的話，控制好資源競爭與防止物件狀態被意外修改是非常重要的，而物件狀態的不變性也是較難以保證的。 而在Scala中，我們可以通過複製不可變狀態的資源（即物件，Scala中一切都是物件，連函式、方法也是）的一個副本，再基於Actor的訊息傳送、接收機制進行並行程式設計

3、Actor方法執行順序

1.首先呼叫start()方法啟動Actor

2.呼叫start()方法後其act()方法會被執行

3.向Actor傳送訊息

4、傳送訊息的方式

! 傳送非同步訊息，沒有返回值。

!? 傳送同步訊息，等待返回值。

!! 傳送非同步訊息，返回值是 Future[Any]。

二、Actor實戰

1、例子一

package cn.itcast.actor
//注意導包是scala.actors.Actor
import scala.actors.Actor
​
object MyActor1 extends Actor{
  //重新act方法
  def act(){
    for(i <- 1 to 10){
      println("actor-1 " + i)
      Thread.sleep(2000)
    }
  }
}
​
object MyActor2 extends Actor{
  //重新act方法
  def act(){
    for(i <- 1 to 10){
      println("actor-2 " + i)
      Thread.sleep(2000)
    }
  }
}
​
object ActorTest extends App{
  //啟動Actor
  MyActor1.start()
  MyActor2.start()
}
 

說明：上面分別呼叫了兩個單例物件的start()方法，他們的act()方法會被執行，相同與在java中開啟了兩個執行緒，執行緒的run()方法會被執行

注意：這兩個Actor是並行執行的，act()方法中的for迴圈執行完成後actor程式就退出了

2、例子二（可以不斷地接收訊息）

class MyActor extends Actor {
​
  override def act(): Unit = {
    while (true) {
      receive {
        case "start" => {
          println("starting ...")
          Thread.sleep(5000)
          println("started")
        }
        case "stop" => {
          println("stopping ...")
          Thread.sleep(5000)
          println("stopped ...")
        }
      }
    }
  }
}
​
object MyActor {
  def main(args: Array[String]) {
    val actor = new MyActor
    actor.start()
    actor ! "start"
    actor ! "stop"
    println("訊息傳送完成！")
  }
}
 

說明：在act()方法中加入了while (true) 迴圈，就可以不停的接收訊息

注意：傳送start訊息和stop的訊息是非同步的，但是Actor接收到訊息執行的過程是同步的按順序執行

3、例子三（react方式會複用執行緒，比receive更高效）

class YourActor extends Actor {
​
  override def act(): Unit = {
    loop {//一個核心一個執行緒，如果一個核心多個執行緒，執行緒還是會切換
      react {
        case "start" => {
          println("starting ...")
          Thread.sleep(5000)
          println("started")
        }
        case "stop" => {
          println("stopping ...")
          Thread.sleep(8000)
          println("stopped ...")
        }
      }
    }
  }
}
​
​
object YourActor {
  def main(args: Array[String]) {
    val actor = new YourActor
    actor.start()
    actor ! "start"
    actor ! "stop"
    println("訊息傳送完成！")
  }
}

說明： react 如果要反覆執行訊息處理，react外層要用loop，不能用while

4、例子四（結合case class傳送訊息）

class AppleActor extends Actor {
​
  def act(): Unit = {
    while (true) {
      receive {
        case "start" => println("starting ...")
        case SyncMsg(id, msg) => {
          println(id + ",sync " + msg)
          Thread.sleep(5000)
          sender ! ReplyMsg(3,"finished")
        }
        case AsyncMsg(id, msg) => {
          println(id + ",async " + msg)
          Thread.sleep(5000)
        }
      }
    }
  }
}
​
object AppleActor {
  def main(args: Array[String]) {
    val a = new AppleActor
    a.start()
    //非同步訊息
    a ! AsyncMsg(1, "hello actor")
    println("非同步訊息傳送完成")
    //同步訊息
    //val content = a.!?(1000, SyncMsg(2, "hello actor"))
    //println(content)
    val reply = a !! SyncMsg(2, "hello actor")
    println(reply.isSet)
    //println("123")
    val c = reply.apply()
    println(reply.isSet)
    println(c)
  }
}
case class SyncMsg(id : Int, msg: String)
case class AsyncMsg(id : Int, msg: String)
case class ReplyMsg(id : Int, msg: String)

三、練習

        用actor併發程式設計寫一個單機版的WorldCount，將多個檔案作為輸入，計算完成後將多個任務彙總，得到最終的結果

class Task extends Actor {
​
  override def act(): Unit = {
    loop {
      react {
        case SubmitTask(fileName) => {
          val contents = Source.fromFile(new File(fileName)).mkString
          val arr = contents.split("\r\n")
          val result = arr.flatMap(_.split(" ")).map((_, 1)).groupBy(_._1).mapValues(_.length)
          //val result = arr.flatMap(_.split(" ")).map((_, 1)).groupBy(_._1).mapValues(_.foldLeft(0)(_ + _._2))
          sender ! ResultTask(result)
        }
        case StopTask => {
          exit()
        }
      }
    }
  }
}
​
object WorkCount {
  def main(args: Array[String]) {
    val files = Array("c://words.txt", "c://words.log")
​
    val replaySet = new mutable.HashSet[Future[Any]]
    val resultList = new mutable.ListBuffer[ResultTask]
​
    for(f <- files) {
      val t = new Task
      //執行緒執行完後，返回結果放到Future[Any]，即返回結果型別為Future[Any]
      val replay = t.start() !! SubmitTask(f)
      replaySet += replay
    }
​
    while(replaySet.size > 0){
      val toCumpute = replaySet.filter(_.isSet)
      for(r <- toCumpute){
        val result = r.apply()//apply，把結果取出來
        resultList += result.asInstanceOf[ResultTask]
        replaySet.remove(r)
      }
      Thread.sleep(100)
    }
    val finalResult = resultList.map(_.result).flatten.groupBy(_._1).mapValues(x => x.foldLeft(0)(_ + _._2))
    println(finalResult)
  }
}
​
case class SubmitTask(fileName: String)
case object StopTask
case class ResultTask(result: Map[String, Int])

注：我們現在學的Scala Actor是scala 2.10.x版本及以前版本的Actor。Scala在2.11.x版本中將Akka加入其中，作為其預設的Actor，老版本的Actor已經廢棄[不可用]