9 Actor介绍
9.1 Java并发编程的问题
在Java并发编程中，每个对象都有一个逻辑监视器（monitor），可以用来控制对象的多线程访问。我们添加sychronized关键字来标记，需要进行同步加锁访问。这样，通过加锁的机制来确保同一时间只有一个线程访问共享数据。但这种方式存在资源争夺、以及死锁问题，程序越大问题越麻烦。
线程死锁
9.2 Actor并发编程模型
Actor并发编程模型，是scala提供给程序员的一种与Java并发编程完全不一样的并发编程模型，是一种基于事件模型的并发机制。Actor并发编程模型是一种不共享数据，依赖消息传递的一种并发编程模式，有效避免资源争夺、死锁等情况。
9.3 Java并发编程对比Actor并发编程
| Java内置线程模型 | scala Actor模型 |
| "共享数据-锁"模型 (share data and lock) | share nothing |
| 每个object有一个monitor，监视线程对共享数据的访问 | 不共享数据，Actor之间通过Message通讯 |
| 加锁代码使用synchronized标识 | |
| 死锁问题 |
| 每个线程内部是顺序执行的 | 每个Actor内部是顺序执行的 |
注：scala在2.11.x版本中加入了Akka并发编程框架，老版本已经废弃。Actor的编程模型和Akka很像，我们这里学习Actor的目的是为学习Akka做准备。
10 Actor编程案例
10.1 创建Actor
创建Actor的方式和Java中创建线程很类似。下面是具体的步骤：
- 定义class或object继承Actor特质
- 重写act方法
- 调用Actor的start方法执行Actor
示例1：创建两个Actor，一个Actor打印1-10，另一个Actor打印11-20
// 继承Actor特质 object Actor1 extends Actor { // 实现act方法 override def act(): Unit = (1 to 10).foreach{ num => println(s"线程1: $num") } } object Actor2 extends Actor { override def act(): Unit = (11 to 20).foreach{ num => println(s"线程2: $num") } } object ActorDemo { def main(args: Array[String]): Unit = { // 调用start启动Actor Actor1.start() Actor2.start() } }
上述代码，分别调用了单例对象的start方法（object本质也是一个class，看成是都包含静态成员的类），会在JVM中开启两个线程来执行act中的代码。
Actor的执行顺序
- 调用start()方法启动Actor
- 自动执行act()方法
- 向Actor发送消息
- act方法执行完成后，程序会调用**exit()**方法
- Actor是并行执行的
- act方法执行完后，Actor程序就退出了
10.2 发送消息/接收消息
我们之前介绍Actor的时候，说过Actor是基于事件（消息）的并发编程模型，那么Actor是如何发送消息和接收消息的呢？
我们可以使用三种方式来发送消息：
| ！ | 发送异步消息，没有返回值 |
| !? | 发送同步消息，等待返回值 |
| !! | 发送异步消息，返回值是Future[Any] |
例如：
要给actor1发送一个异步字符串消息，使用以下代码：
actor1 ! "你好!"
示例：创建两个Actor，Actor1发送一个异步字符串消息给Actor2，Actor2接收到该消息后，打印出来。
object ActorSender extends Actor { override def act(): Unit = { // 发送字符串消息给Actor2 val msg = "你好，ActorSender" println(s"ActorSender: 发送消息$msg") ActorReceiver ! msg } } object ActorReceiver extends Actor { override def act(): Unit = receive { case msg: String => println(s"接收Actor: 接收到$msg") } } object ActorMsgDemo { def main(args: Array[String]): Unit = { ActorSender.start() ActorReceiver.start() } }
- 使用!、!?、!!来发送消息
- actor中使用receive方法来接收消息，需要给receive方法传入一个偏函数
{ case 变量名1:消息类型1 => 业务处理1, case 变量名2:消息类型2 => 业务处理2, ... }
- receive方法只接收一次消息，接收完后结束Actor
10.3 持续接收消息
上一个案例，ActorReceiver调用receive来接收消息，但接收一次后，Actor就退出了。
object ActorSender extends Actor { override def act(): Unit = { // 发送字符串消息给Actor2 val msg = "你好，ActorSender" println(s"ActorSender: 发送消息$msg") ActorReceiver ! msg // 再次发送一条消息，ActorReceiver无法接收到 ActorReceiver ! "你叫什么名字？" } } object ActorReceiver extends Actor { override def act(): Unit = receive { case msg: String => println(s"接收Actor: 接收到$msg") } } object ActorMsgDemo { def main(args: Array[String]): Unit = { ActorSender.start() ActorReceiver.start() } }
上述代码，ActorReceiver无法接收到ActorSender发送的第二条消息。
我们希望ActorReceiver能够接收多条消息，怎么实现呢？
——我们只需要使用一个while(true)循环，不停地调用receive来接收消息就可以啦。
object ActorSender extends Actor { override def act(): Unit = { // 发送字符串消息给Actor2 val msg = "你好，ActorSender" println(s"ActorSender: 发送消息$msg") ActorReceiver ! msg // 再次发送一条消息，ActorReceiver无法接收到 ActorReceiver ! "你叫什么名字？" } } object ActorReceiver extends Actor { override def act(): Unit = // 使用while循环不停地接收消息 while(true) { receive { case msg: String => println(s"接收Actor: 接收到$msg") } } } object ActorMsgDemo { def main(args: Array[String]): Unit = { ActorSender.start() ActorReceiver.start() } }
10.4 共享线程
上述代码，使用while循环来不断接收消息。但如果当前Actor没有接收到消息，线程就会处于阻塞状态。考虑如果有很多的Actor，就有可能会导致很多线程都是处于阻塞状态。这样当有新的消息来时，需要重新创建线程来处理。这样会导致频繁的线程创建、销毁和切换，从而影响运行效率。
在scala中，可以使用loop + react来复用线程。比while + receive更高效
示例：使用loop + react重写上述案例
loop { react{ case msg:String => println(s"接收Actor: 接收到$msg") } }
10.5 发送和接收自定义消息
我们前面发送的消息是字符串类型，Actor中也支持发送自定义消息，常见的如：使用样例类封装消息，然后进行发送处理。
案例：
每一种消息使用样例类来封装，每一个消息有一个id:Int、msgString成员
示例：
object MessageActor extends Actor { override def act(): Unit = loop { react{ // 接收同步消息 case SyncMessage(id, msg) => println(s"接收到SyncMessage消息：id=${id}, msg=${msg}") // 使用sender来获取发送方Actor的引用 sender ! ReplyMessage(2, "回复消息") case ASyncWithoutMessage(id, msg) => println(s"接收到ASyncWithoutMessage消息：id=${id}, msg=${msg}") case ASyncWithMessage(id, msg) => println(s"接收到ASyncMessage消息：id=${id}, msg=${msg}") // 3秒后再回复消息 TimeUnit.SECONDS.sleep(3) sender ! ReplyMessage(5, "回复消息") case _ => println("未知消息") } } } object MainActor { def main(args: Array[String]): Unit = { MessageActor.start() // 发送同步消息 println("发送同步消息") println("-----") val replyMessage = MessageActor !? SyncMessage(1, "同步消息") println(replyMessage) println("-----") // 发送异步无返回消息 println("发送异步无返回消息") MessageActor ! ASyncWithoutMessage(3, "异步无返回消息") // 睡眠两秒，等待Actor接收到并打印异步消息，方便观察测试 TimeUnit.SECONDS.sleep(2) println("-----") // 发送异步有返回消息 println("发送异步有返回消息") val future: Future[Any] = MessageActor !! ASyncWithMessage(4, "异步有返回消息") // 这个Future表示对返回消息的封装，因为发的异步消息，所以不确定在将来哪个时间会返回消息 // 使用循环，不断调用isSet来检查是否已经接收到消息 while(!future.isSet) { } val asyncReplyMessage = future.apply() println(s"接收到ReplyMessage: $asyncReplyMessage") } }
- 在编写Actor程序时，一般使用样例类来封装消息
- 在Actor的act方法中，可以使用sender来获取发送方Actor的引用
- Future表示发送有返回的异步消息的封装，虽然获取到了Future的返回值，但Future中不一定有值，因为可能在将来的某一时刻才会返回消息
- 使用Future的isSet()可以检查是否已经收到返回消息，使用apply()方法可以获取返回的消息。
- 使用TimeUnit.SECONDS.sleep来让Actor对应的线程睡眠阻塞
10.6 基于Actor实现WordCount案例
接下来，我们要使用Actor并发编程模型实现多文件的单词统计。
实现步骤：
- MainActor获取要进行单词统计的文件
- 根据文件数量创建对应的WordCountActor
- 将文件名封装为消息发送给WordCountActor
- WordCountActor接收消息，并统计单个文件的单词计数
- 将单词计数结果回复给MainActor
- MainActor等待所有的WordCountActor都已经成功返回消息，然后进行结果合并
代码：
case class WordCountTask(fileName:String) case class WordCountResultMessage(wcResult:Map[String,Int]) class WordCountActor extends Actor { override def act(): Unit = { loop { react { // 接收任务，获取文件名 case WordCountTask(fileName) => val actorId = this.toString.split("@")(1) println(s"${actorId}接收到任务: $fileName") // 读取文件 val lineArr: Array[String] = Source.fromFile(fileName).mkString.split("\r\n") val wordArr: Array[String] = lineArr.flatMap(_.split(" ")) val wordnumArr: Array[(String, Int)] = wordArr.map(_->1) val groupedWordNumMap: Map[String, Array[(String, Int)]] = wordnumArr.groupBy(_._1) val result: Map[String, Int] = groupedWordNumMap.mapValues(_.foldLeft(0)(_ + _._2)) println(s"${actorId}返回结果：${result.toList.sorted}") sender ! WordCountResultMessage(result) } } } } object WordCountDemo { def main(args: Array[String]): Unit = { val dir = "./data/textfiles" val files = List(s"$dir/1.txt", s"$dir/2.txt", s"$dir/3.txt") // 启动若干个WordCountActor val wordCountActors: List[(String, WordCountActor)] = files.map { file => val wordCountActor = new WordCountActor() wordCountActor.start() (file, wordCountActor) } // 发送文件名给每一个启动的Actor val futureList: List[Future[Any]] = wordCountActors.map { fileAndActor => // 发送WordCountTask给Actor fileAndActor._2 !! WordCountTask(fileAndActor._1) } // 等待所有结果读已经返回 while(futureList.count(!_.isSet) != 0) {} // 获取所有结果 val wordCountResults: List[(String, Int)] = futureList.flatMap(_.apply().asInstanceOf[WordCountResultMessage].wcResult.toList) // 按照单词进行分组 val groupWordCountResults: Map[String, List[(String, Int)]] = wordCountResults.groupBy(_._1) val finalResult: Map[String, Int] = groupWordCountResults.mapValues(_.foldLeft(0)(_ + _._2)) println("-" * 10) println("合并后结果为：") println(finalResult.toList.sorted) } }
