溫馨提示×

溫馨提示×

您好，登錄后才能下訂單哦！

密碼登錄×
登錄注冊×
其他方式登錄
點(diǎn)擊 登錄注冊 即表示同意《億速云用戶服務(wù)條款》

好程序員大數(shù)據(jù)學(xué)習(xí)路線分享Actor學(xué)習(xí)筆記

發(fā)布時(shí)間：2020-05-24 15:22:56 來源：網(wǎng)絡(luò) 閱讀：204 作者：wx5d42865f47214 欄目：大數(shù)據(jù)

　　好程序員大數(shù)據(jù)學(xué)習(xí)路線分享Actor學(xué)習(xí)筆記，在scala中她能實(shí)現(xiàn)很強(qiáng)大的功能,他是基于并發(fā)機(jī)制的一個(gè)事件模型

我們現(xiàn)在學(xué)的scala2.10.x版本就是之前的Actor

?

同步:在主程序上排隊(duì)執(zhí)行的任務(wù),只有前一個(gè)任務(wù)執(zhí)行完畢后,才能執(zhí)行下一個(gè)任務(wù)

異步:指不進(jìn)入主程序,而進(jìn)入"任務(wù)對列"的任務(wù),只有等主程序任務(wù)執(zhí)行完畢,"任務(wù)對列"開始請求主程序,請求任務(wù)執(zhí)行,該任務(wù)會(huì)進(jìn)入主程序

?

java

共享變量 -- 加鎖

會(huì)出現(xiàn)鎖死問題

scala

Actor不共享數(shù)據(jù)

沒有鎖的概念

Actor通信之間需要message(通信)

Aactor執(zhí)行順序

1.首先調(diào)用start()方法啟動(dòng)Actor

2.調(diào)用start()方法后act()方法會(huì)被執(zhí)行

3.Actor之間進(jìn)行發(fā)送消息

Actor發(fā)送消息的三種方式

! -> 發(fā)送異步消息,沒有返回值

!? -> 發(fā)送同步消息,有返回值,會(huì)有線程等待

!! -> 發(fā)送異步消息,有返回值,返回值類型Future[Any](用來獲取異步操作結(jié)果)

?

Actor并行執(zhí)行

//注意,這兩個(gè)actor會(huì)并行執(zhí)行,當(dāng)其中一個(gè)for循環(huán)結(jié)束后,actor結(jié)束

object ActorDemo01 {

??def main(args: Array[String]): Unit = {

????MyActor1.start()

????MyActor2.start()

??}

}

?

object MyActor1 extends Actor{

??override def act(): Unit = {

????for (i <- 1 to 10){

??????println(s"actor => $i")

??????Thread.sleep(2000)

????}

??}

?

??object MyActor2 extends Actor{

????override def act(): Unit = {

??????for (i <- 1 to 5){

????????println(s"actor2 => $i")

????????Thread.sleep(2000)

??????}

????}

??}

}

用Actor不斷接受消息

執(zhí)行第一種方式,異步

object ActorDemo02 {

??def main(args: Array[String]): Unit = {

????val actor: MyActor = new MyActor

????actor.start()

?

????//并行執(zhí)行

????actor ! "start" ?// !->異步

????actor ! "stop"

????println("發(fā)送完成")

?

??}

}

?

class MyActor extends Actor{

??override def act(): Unit = {

????while (true){ ??//死循環(huán)

??????receive { ??//接收

????????case "start" => {

??????????println("starting")

??????????Thread.sleep(1000)

??????????println("started")

????????}

????????case "stop" => {

??????????println("stopping")

??????????Thread.sleep(1000)

??????????println("stopped")

????????}

??????}

????}

??}

}

第二種方式:利用react來代替receive,也就是說react線程可復(fù)用,比receive更高效

object ActorDemo03 {

??def main(args: Array[String]): Unit = {

????val actor: MyActor3 = new MyActor3

????actor.start()

????actor ! "start"

????actor ! "stop"

????println("成功了")

??}

}

?

class MyActor3 extends Actor{

??override def act(): Unit = {

????loop {

??????react{

????????case "start" =>{

??????????println("starting")

??????????Thread.sleep(1000)

??????????println("sarted")

????????}

????????case "stop" =>{

??????????println("stoppting")

??????????Thread.sleep(1000)

??????????println("stopped")

????????}

??????}

????}

??}

}

結(jié)合樣例類練習(xí)Actor發(fā)送消息

//創(chuàng)建樣例類

case class AsyncMsg(id: Int, msg: String)

case class SyncMsg(id: Int, msg: String)

case class ReplyMsg(id: Int, msg: String)

?

object ActorDemo01 extends Actor {

??override def act(): Unit = {

????while (true) {

??????receive {

????????case "start" => println("starting...")

????????case AsyncMsg(id, msg) =>

????????{

??????????println(s"id:$id,msg:$msg")

??????????sender ! ReplyMsg(1,"sucess") ?//接收到消息后返回響應(yīng)消息

????????}

????????case SyncMsg(id,msg) => {

??????????println(s"id:$id,msg:$msg")

??????????sender ! ReplyMsg(2,"sucess")

????????}

??????}

????}

??}

}

?

object ActorTest{

??def main(args: Array[String]): Unit = {

?

????val actor: Actor = ActorDemo01.start()

?

// ???//異步發(fā)送消息,沒有返回值

// ???actor ! AsyncMsg(3,"heihei")

// ???println("異步消息發(fā)送完成,沒有返回值")

?

// ???//同步發(fā)送消息,有返回值

// ???val text: Any = actor !? SyncMsg(4,"OK")

// ???println(text)

// ???println("同步消息發(fā)送成功")

?

????//異步發(fā)送消息,有返回值,返回類型為Future[Any]

????val reply: Future[Any] = actor !! SyncMsg(5,"OK is 不存在的")

????Thread.sleep(2000)

????if (reply.isSet){

??????val applyMsg: Any = reply.apply()

??????println(applyMsg)

????}else{

??????println("Nothing")

????}

??}

}

Actor并行化的wordcount

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

??????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)

??????}

??????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])

?


向AI問一下細(xì)節(jié)

免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場，如果涉及侵權(quán)請聯(lián)系站長郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。

AI
助
手