Scala中的Actor模式

1. Actor介绍

scala中的Actor并发编程模型可以用来开发比java线程效率更高的并发程序

2. JAVA并发编程的问题

在Java并发编程中,每个对象都有一个逻辑监视器(monitor),可以用来控制对象的多线程访问,我们添加sychronized关键字来标记,需要进行同步加锁访问.这样,通过加锁的机制来确保同一时间只能有一个线程访问共享数据,但是这种方式存在资源争夺,以及死锁的问题,程序越大问题越麻烦

线程死锁  

3. Acator并发编程模型

Actor并发编程模型,是scala提供给程序员的一种与java并发编程完全不一样的并发编程模型,是一种基于事件模型的并发机制,Actor并发编程模型是一种不共享数据,依赖消息传递的一种并发编程模式,有效避免资源争夺,死锁等情况

4.Java并发编程对比Actor并发编程

Java内置线程模型scala Actor模型"共享数据-锁"模型share nothing每个object有一个monitor,监视线程堆共享数据的访问不共享数据,Actor之间通过Message通讯加锁代码使用synchronized标识死锁问题每个线程内部都是顺序执行的每个Actor内部是顺序执行的

5. Actor的简单使用

创建Actor的方式和java中创建线程很相似,也是通过继承来创建

5.1 使用方式

定义class或者object来继承Actor特质重写act方法调用Actor的start方法执行Actor

示例:

创建两个Actor,一个Actor打印1-10,另一个Actor打印11-20

使用class继承Actor创建(程序中需要创建多个相同的Actor)使用object继承Actor创建(程序中只创建一个Actor)

参考代码 使用class继承Actor创建

class Actor1 extends Actor{ override def act(): Unit = (1 to 10).foreach(println) } class Actor2 extends Actor{ override def act(): Unit = (11 to 20).foreach(println) } def main(args: Array[String]): Unit = { new Actor1().start() new Actor2().start() }

使用object继承Actor创建

object Actor1 extends Actor{ override def act(): Unit = (1 to 10).foreach(println) } object Actor2 extends Actor{ override def act(): Unit = (11 to 20).foreach(println) } def main(args: Array[String]): Unit = { Actor1.start() Actor2.start() }

5.2 Actor执行流程

调用start()方法启动Actor自动执行**act()**方法向Actor发送消息act方法执行完成后,程序会调用**exit()**方法

5.3 发送消息/接收消息

5.3.1 发送消息

三种方式

!发送异步消息,没有返回值!?发送同步消息,等待返回值!!发送异步消息,返回值是Future[Any]

例如: 要给actor1发送一个异步字符串消息,可以这样玩:

actor1 ! "你好!"

5.3.2 接收消息

Actor中使用receive方法来接收消息,需要给receive方法传入一个偏函数

{ case 变量名1:消息类型1=>业务处理1, case 变量名2:消息类型2=>业务处理2, ... }

注意:receive方法值接收一次消息,接收完后继续执行act方法

示例:

创建两个Actor(ActorSender,ActorReceiver)ActorSender发送一个异步字符串消息给ActorReceiverActorReceive接收到该消息后,打印出来 参考代码:

receive来接收消息,只接收一次后,Actor就退出了,可以使用while(true)循环,不停的调用receive来接收消息

import java.util.concurrent.TimeUnit import scala.actors.Actor object Demo16 { object ActorSender extends Actor{ override def act(): Unit = { while(true){ ActorReceiver ! "今天天气不错..............." TimeUnit.SECONDS.sleep(3)//休眠时间 } } } object ActorReceiver extends Actor{ override def act(): Unit = { while (true){ receive{ case msg:String=>println("接收到消息:"+msg) } } } } def main(args: Array[String]): Unit = { ActorSender.start() ActorReceiver.start() } }

5.4 使用loop和react优化接收消息

使用while循环来不断接收消息:

如果Actor没有接收到消息,线程就会处于阻塞状态如果有很多的Actor,就有可能会导致很多线程处于阻塞状态每次有新的消息来时,都得重新创建线程来处理频繁的线程创建,销毁和切换,都会影响运行效率的 由于while(true)方式复用线程有诸多缺点,因此可以使用loop+recat来优化

示例: 使用loop+react来重写上面的案例:

import java.util.concurrent.TimeUnit import scala.actors.Actor object Demo17 { object ActorSender extends Actor{ override def act(): Unit = { loop{ ActorReceiver ! "今天天气不错" TimeUnit.SECONDS.sleep(3)//休眠时间 } } } object ActorReceiver extends Actor{ override def act(): Unit = { loop{ react{ case msg:String=>println("接收到消息:"+msg) } } } } def main(args: Array[String]): Unit = { ActorSender.start() ActorReceiver.start() } }

5.4 发送和接收自定义消息

上诉案例演示发送的消息都是字符串类型,Actor中也支持自定义消息:使用样例类封装消息,然后进行发送处理 示例一: [同步有返回值*]

创建一个MsgActor，并向它发送一个同步消息，该消息包含两个字段（id、message）MsgActor回复一个消息，该消息包含两个字段（message、name）打印回复消息

注意:

使用 !? 发送同步返回值消息在Actor的act方法中,可以使用sender获取发送者的Actor引用 参考代码: import scala.actors.Actor object Demo19 { case class Message(id:Int,msg:String) case class ReplyMessage(msg:String,name:String) object MsgActor extends Actor{ override def act(): Unit = { loop{ react{ case Message(id,msg)=>println(s"接收到消息:${id},${msg}") sender ! ReplyMessage("不太好","Tom") } } } } def main(args: Array[String]): Unit = { MsgActor.start() val replyMessage:Any = MsgActor !? Message(1,"你好") println("回复消息:"+replyMessage.asInstanceOf[ReplyMessage]) } }

示例二: [异步无返回值]

创建一个MsgActor,并向它发送一个异步无返回值的消息,该消息包含了两字段(message,company)

注意:

使用 ! 发送异步无返回消息 参考代码: import scala.actors.Actor object Demo20 { case class Message(message:String,company:String) object MsgActor extends Actor{ override def act(): Unit = { loop{ react{ case Message(message,company)=>println(s"MsgActor接收到消息:${message},${company}") } } } } def main(args: Array[String]): Unit = { MsgActor.start() MsgActor ! Message("中国联通","话费充满") } }

示例三: [异步有返回值]

创建一个MsgActor，并向它发送一个异步有返回消息，该消息包含两个字段（id、message）MsgActor回复一个消息，该消息包含两个字段（message、name）打印回复消息

注意:

使用!! 发送异步有返回消息发送后,返回类型为Future[Any]对象Future表示异步返回数据的封装,虽然获取到了Future的返回值,但是不一定有值,可能在某一时候才会返回消息Future的isSet()可检查是否已将收到返回消息,apply()方法可获取返回数据

参考代码:

import scala.actors.Actor object Demo18 { //定义两个样例类(Message发送消息,ReplyMessage表示返回的数据) case class Message(id:Int,message:String) case class ReplyMessage(message:String,name:String) //用来接收MainActor发送的消息,并且回复一条消息 object MyActor extends Actor { override def act(): Unit = { loop{ react{ case Message(id,message)=>println(s"我是Message,接收到的消息是:${id},${message}") sender ! ReplyMessage("我是ReplyMessage,接收到了你的消息","糖糖") } } } } def main(args: Array[String]): Unit = { MyActor.start() //采用异步有返回的形式发送消息 val future = MyActor !! Message(1,"你好,我是Message,我在给你发消息") //Future的isSet()可检测是否收到消息,apply()获取返回数据 while (!future.isSet){ val result = future.apply().asInstanceOf[ReplyMessage] println(result) } } }