23 Actor 并发编程模型
23.1 Actor概述
为什么要学习actor?
我们现在学的Scala Actor是Scala 2.10.x版本及以前版本的Actor；
Scala在2.11.x版本中将Akka加入其中，作为默认的Actor，老版本的Actor已经废弃；
我们学习Actor的目的就是为了学习Akka做铺垫；
什么是Actor？
Actor是消息并发模型；
Scala中的Actor能够实现并行编程的强大功能，它是基于事件模型的并发机制。
Scala是运用消息（message）的发送、接收来实现多线程的。
使用Scala能够更容易地实现多线程应用的开发。
Actor是一个封装了状态和行为的对象，Actor之间可以通过交换消息的方式进行通信，每个Actor都有自己的收件箱（MailBox）。
它每次只能处理一条消息，所以actor内部可以安全的处理状态，而不用考虑锁机制。
Java并发编程与Scala Actor编程的区别
Scala的Actor类似于Java中的多线程编程，但是不同的是，Scala的Actor提供的模型与多线程有所不同。
Scala的Actor尽可能地避免锁和共享状态，从而避免多线程并发时出现资源争用的情况。进而提升多线程编程的性能。
此外，Scala Actor的这种模型还可以避免死锁等一系列传统多线程编程的问题。
原因就在于Java中多数使用的是可变状态的对象资源，对这些资源进行共享来实现多线程编程的话，
控制好资源竞争与防止对象状态被意外修改是非常重要的，而对象状态的不变性也是较难以保证的。
而在Scala中，我们可以通过复制不可变状态的资源（即对象，Scala中一切都是对象，连函数、方法也是）的一个副本，
再基于Actor的消息发送、接收机制进行并行编程；
Actor方法执行顺序
1）调用 start() 方法启动 Actor；
2）执行 act() 方法；
3）向 Actor 发送消息；
发送消息的方式
! 发送异步消息，没有返回值；
!? 发送同步消息，等待返回值；
!! 发送异步消息，返回值是Future[Any]；
23.2 actor 使用
scala 的 actor 在 scala 高版本被抛弃，如果想用，需要在工程中加入额外的 jar 包
23.2.1 actor 简单实现
package day05
import scala.actors.Actor
// 相当于继承 Thread
class ActorDemo extends Actor {
// run方法体
// 一个线程只执行一次
override def act(): Unit = {
println("一个线程只执行一次")
}
}
object ActorDemo{
def main(args: Array[String]): Unit = {
val demo = new ActorDemo
// 启动线程
demo.start()
}
}
23.2.2 actor通信
package day05
import scala.actors.{Actor, Future}
// 相当于继承 Thread
class ActorDemo extends Actor {
// run方法体
// 一个线程只执行一次
override def act(): Unit = {
println("一个线程只执行一次")
while(true){
receive{
case "hello" => println("actor receive==> hello")
case "id01" =>{
println("actor receive==> id01, sleep 5s, return id02")
// Thread.sleep(5000)
// 回消息
sender ! "id02"
}
case AsyncSendAndReturnMsg(id) => {
println(s"actor receive==> ${id}, return id04")
Thread.sleep(3000)
sender ! AsyncSendAndReturnMsg("id04")
}
}
}
}
}
object ActorDemo{
def main(args: Array[String]): Unit = {
val demo = new ActorDemo
// 启动线程
demo.start()
// 发送异步无返回消息
demo ! "hello"
// 发送同步有返回消息
// 同步效率低
val res1: Any = demo !? "id01"
println(s"同步返回消息：${res1.asInstanceOf[String]}")
// 异步效率高
// 发送异步有返回消息
val future: Future[Any] = demo !! AsyncSendAndReturnMsg("id03")
println("继续干活，耗时4s，此时是能得到actor返回的结果")
Thread.sleep(4000)
// println("继续干活，耗时1s，得不到actor返回的结果")
// Thread.sleep(1000)
if(future.isSet){
// future() 底层调用apply方法
val res2: Any = future()
val msg = res2.asInstanceOf[AsyncSendAndReturnMsg]
println(msg)
}else{
println("future 没有等到结果")
}
}
}
// 用于封装发送和返回 异步消息
case class AsyncSendAndReturnMsg(val id:String)
23.2.3 用actor实现wordcount
package day05
import scala.actors.{Actor, Future}
import scala.collection.mutable.ListBuffer
import scala.io.Source
class WordCountActor extends Actor{
override def act(): Unit = {
receive{
case path:String =>{
println(s"actor receive==> path:${path}, start ...")
val list = Source.fromFile(path).getLines().toList
val map: Map[String, Int] = list.flatMap(_.split("\t")).map((_,1)).groupBy(_._1).mapValues(_.size)
sender ! map
}
}
}
}
object WordCountActor{
def main(args: Array[String]): Unit = {
val files = Array("/tmp/scala/input/word1.txt",
"/tmp/scala/input/word2.txt",
"/tmp/scala/input/word3.txt",
"/tmp/scala/input/word4.txt")
// 用来存future的
val futures = new ListBuffer[Future[Any]]
// 用来装从future里提取出来的map数据
val maps = new ListBuffer[Map[String,Int]]
// 为每个文件创建一个actor对象处理wordcount
for(f <- files){
val actor = new WordCountActor
actor.start()
val future: Future[Any] = actor !! f
futures += future
}
// 循环判断futures列表中的Futrue是否有数据返回，如果有，提取里面的数据
while(! futures.isEmpty){
val hasDataFutures = futures.filter(_.isSet)
for(hasDataFutrue <- hasDataFutures){
// 提取数据
val data: Any = hasDataFutrue()
val map = data.asInstanceOf[Map[String,Int]]
println(s"局部map：${map}")
maps += map
// 从future列表里，把已经提取出数据的Futrue干掉
futures -= hasDataFutrue
}
}
// 把多个结果的map合并成一个
val list1: ListBuffer[(String, Int)] = maps.flatten
val resMap: Map[String, Int] = list1.groupBy(_._1).mapValues(_.map(_._2).sum)
val resList: List[(String, Int)] = resMap.toList.sortBy(_._2).reverse
println(s"resList: ${resList}")
}
}
运行结果：
海汼部落原创文章，原文链接：http://hainiubl.com/topics/75754
