obj submit for 什麽 資源競爭 運用 art http 線程

1.1. 課程目標
1.1.1. 目標一：熟悉Scala Actor並發編程
1.1.2. 目標二：為學習Akka做準備
註：Scala Actor是scala 2.10.x版本及以前版本的Actor。
Scala在2.11.x版本中將Akka加入其中，作為其默認的Actor，老版本的Actor已經廢棄。
1.2. 什麽是Scala Actor1.2.1. 概念
Scala中的Actor能夠實現並行編程的強大功能，它是基於事件模型的並發機制，Scala是運用消息的發送、接收來實現高並發的。
Actor可以看作是一個個獨立的實體，他們之間是毫無關聯的。但是，他們可以通過消息來通信。一個Actor收到其他Actor的信息後，它可以根據需要作出各種相應。消息的類型可以是任意的，消息的內容也可以是任意的。
1.2.2. java並發編程與Scala Actor編程的區別

對於Java，我們都知道它的多線程實現需要對共享資源（變量、對象等）使用synchronized 關鍵字進行代碼塊同步、對象鎖互斥等等。而且，常常一大塊的try…catch語句塊中加上wait方法、notify方法、notifyAll方法是讓人很頭疼的。原因就在於Java中多數使用的是可變狀態的對象資源，對這些資源進行共享來實現多線程編程的話，控制好資源競爭與防止對象狀態被意外修改是非常重要的，而對象狀態的不變性也是較難以保證的。
與Java的基於共享數據和鎖的線程模型不同，Scala的actor包則提供了另外一種不共享任何數據、依賴消息傳遞的模型,從而進行並發編程。
1.2.3. Actor的執行順序
1、首先調用start()方法啟動Actor
2、調用start()方法後其act()方法會被執行
3、向Actor發送消息
4、act方法執行完成之後，程序會調用exit方法
1.2.4. 發送消息的方式

註意：Future 表示一個異步操作的結果狀態，可能還沒有實際完成的異步任務的結果
Any 是所有類的超類，Future[Any]的泛型是異步操作結果的類型。
1.3. Actor實戰1.3.1. 第一個例子
怎麽實現actor並發編程：
1、定義一個class或者是object繼承Actor特質，註意導包import scala.actors.Actor
2、重寫對應的act方法
3、調用Actor的start方法執行Actor
4、當act方法執行完成，整個程序運行結束

package cn.itcast.actor

import scala.actors.Actor

object Actor1 extends Actor{
  //重寫act方法
  def act(){
    for(i <- 1 to 10){
      println("actor-1 " + i)
      Thread.sleep(2000)
    }
  }
}

object Actor2 extends Actor{
  //重寫act方法
  def act(){
    for(i <- 1 to 10){
      println("actor-2 " + i)
      Thread.sleep(2000)
    }
  }
}

object ActorTest extends App{
  //啟動Actor
  Actor1.start()
  Actor2.start()
}

說明：上面分別調用了兩個單例對象的start()方法，他們的act()方法會被執行，相同與在java中開啟了兩個線程，線程的run()方法會被執行
註意：這兩個Actor是並行執行的，act()方法中的for循環執行完成後actor程序就退出了
1.3.2. 第二個例子
怎麽實現actor發送、接受消息
1、定義一個class或者是object繼承Actor特質，註意導包import scala.actors.Actor
2、重寫對應的act方法
3、調用Actor的start方法執行Actor
4、通過不同發送消息的方式對actor發送消息
5、act方法中通過receive方法接受消息並進行相應的處理
6、act方法執行完成之後，程序退出

package cn.itcast.actor
import scala.actors.Actor
class MyActor extends Actor {

  override def act(): Unit = {
      receive {
        case "start" => {
          println("starting ...")
        }
      }
    }
  }
}

object MyActor {
  def main(args: Array[String]) {
    val actor = new MyActor
    actor.start()
    actor ! "start"

    println("消息發送完成！")

  }
}

1.3.3. 第三個例子
怎麽實現actor可以不斷地接受消息：
在act方法中可以使用while(true)的方式，不斷的接受消息。

package cn.itcast.actor
import scala.actors.Actor
class MyActor1 extends Actor {

  override def act(): Unit = {
    while (true) {
      receive {
        case "start" => {
          println("starting ...")
        }
        case "stop" => {
          println("stopping ...")
        }
      }
    }
  }
}

object MyActor1 {
  def main(args: Array[String]) {
    val actor = new MyActor1
    actor.start()
    actor ! "start"
    actor ! "stop"

  }
}

說明：在act()方法中加入了while (true) 循環，就可以不停的接收消息
註意：發送start消息和stop的消息是異步的，但是Actor接收到消息執行的過程是同步的按順序執行
1.3.4. 第四個例子
使用react方法代替receive方法去接受消息
好處：react方式會復用線程，避免頻繁的線程創建、銷毀和切換。比receive更高效
註意: react 如果要反復執行消息處理，react外層要用loop，不能用while

package cn.itcast.actor
import scala.actors.Actor
class YourActor extends Actor {
  override def act(): Unit = {
    loop {
      react {
        case "start" => {
          println("starting ...")
        }
        case "stop" => {
          println("stopping ...")

        }
      }
    }
  }
}

object YourActor {
  def main(args: Array[String]) {
    val actor = new YourActor
    actor.start()
    actor ! "start"
    actor ! "stop"
    println("消息發送完成！")

  }
}

1.3.5. 第五個例子
結合case class樣例類發送消息和接受消息
1、將消息封裝在一個樣例類中
2、通過匹配不同的樣例類去執行不同的操作
3、Actor可以返回消息給發送方。通過sender方法向當前消息發送方返回消息

package cn.itcast.actor
import scala.actors.Actor

case class SyncMessage(id:Int,msg:String)//同步消息
case class AsyncMessage(id:Int,msg:String)//異步消息
case class ReplyMessage(id:Int,msg:String)//返回結果消息

class MsgActor extends Actor{
  override def act(): Unit ={
    loop{
      react{
        case "start"=>{println("starting....")}

        case SyncMessage(id,msg)=>{
          println(s"id:$id, SyncMessage: $msg")
          Thread.sleep(2000)
          sender !ReplyMessage(1,"finished...")
        }
        case AsyncMessage(id,msg)=>{
          println(s"id:$id,AsyncMessage: $msg")
         // Thread.sleep(2000)
          sender !ReplyMessage(3,"finished...")
          Thread.sleep(2000)
        }

      }
    }
  }
}

object MainActor {
  def main(args: Array[String]): Unit = {
     val mActor=new MsgActor
        mActor.start()
        mActor!"start"

        //同步消息 有返回值
     val reply1= mActor!?SyncMessage(1,"我是同步消息")
      println(reply1)
      println("===============================")
        //異步無返回消息
     val reply2=mActor!AsyncMessage(2,"我是異步無返回消息")

      println("===============================")
        //異步有返回消息
    val reply3=mActor!!AsyncMessage(3,"我是異步有返回消息")
    //Future的apply()方法會構建一個異步操作且在未來某一個時刻返回一個值
      val result=reply3.apply()
      println(result)

  }
}

1.3.6. 練習實戰
需求：
用actor並發編程寫一個單機版的WordCount，將多個文件作為輸入，計算完成後將多個任務匯總，得到最終的結果。
大致的思想步驟：
1、通過loop +react 方式去不斷的接受消息
2、利用case class樣例類去匹配對應的操作
3、其中scala中提供了文件讀取的接口Source,通過調用其fromFile方法去獲取文件內容
4、將每個文件的單詞數量進行局部匯總，存放在一個ListBuffer中
5、最後將ListBuffer中的結果進行全局匯總。

package cn.itcast.actor
import java.io.File
import scala.actors.{Actor, Future}
import scala.collection.mutable
import scala.io.Source

case class SubmitTask(fileName: String)
case class ResultTask(result: Map[String, Int])
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)
        }
      }
    }
  }
}

object WorkCount {
  def main(args: Array[String]) {
    val files = Array("d://aaa.txt", "d://bbb.txt","d://ccc.txt")
    val replaySet = new mutable.HashSet[Future[Any]]
    val resultList = new mutable.ListBuffer[ResultTask]
    for(f <- files) {
      val t = new Task
      val replay = t.start() !! SubmitTask(f)
      replaySet += replay
    }
    while(replaySet.size > 0){
      val toCumpute = replaySet.filter(_.isSet)
      for(r <- toCumpute){
        val result = r.apply()
        resultList += result.asInstanceOf[ResultTask]
        replaySet.remove(r)
      }

    }
    val finalResult = resultList.map(_.result).flatten.groupBy(_._1).mapValues(x => x.foldLeft(0)(_ + _._2))
    println(finalResult)
  }
}

spark筆記之Scala演員並發編程