好程序员大数据学习路线分享Actor学习笔记
好程序员大数据学习路线分享Actor学习笔记，在scala中她能实现很强大的功能,他是基于并发机制的一个事件模型
我们现在学的scala2.10.x版本就是之前的Actor
同步:在主程序上排队执行的任务,只有前一个任务执行完毕后,才能执行下一个任务
异步:指不进入主程序,而进入"任务对列"的任务,只有等主程序任务执行完毕,"任务对列"开始请求主程序,请求任务执行,该任务会进入主程序
java
共享变量 -- 加锁
会出现锁死问题
scala
Actor不共享数据
没有锁的概念
Actor通信之间需要message(通信)
Aactor执行顺序
1.首先调用start()方法启动Actor
2.调用start()方法后act()方法会被执行
3.Actor之间进行发送消息
Actor发送消息的三种方式
! -> 发送异步消息,没有返回值
!? -> 发送同步消息,有返回值,会有线程等待
!! -> 发送异步消息,有返回值,返回值类型Future[Any](用来获取异步操作结果)
Actor并行执行
//注意,这两个actor会并行执行,当其中一个for循环结束后,actor结束 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不断接受消息
执行第一种方式,异步
object ActorDemo02 { def main(args: Array[String]): Unit = { val actor: MyActor = new MyActor actor.start() //并行执行 actor ! "start" // !->异步 actor ! "stop" println("发送完成") } } class MyActor extends Actor{ override def act(): Unit = { while (true){ //死循环 receive { //接收 case "start" => { println("starting") Thread.sleep(1000) println("started") } case "stop" => { println("stopping") Thread.sleep(1000) println("stopped") } } } } } |
第二种方式:利用react来代替receive,也就是说react线程可复用,比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") } } } } } |
结合样例类练习Actor发送消息
//创建样例类 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") //接收到消息后返回响应消息 } 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() // //异步发送消息,没有返回值 // actor ! AsyncMsg(3,"heihei") // println("异步消息发送完成,没有返回值") // //同步发送消息,有返回值 // val text: Any = actor !? SyncMsg(4,"OK") // println(text) // println("同步消息发送成功") //异步发送消息,有返回值,返回类型为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]) |
