ScalaActor介绍-阿南达文事网

ScalaActor介绍

文章目录

Actor介绍
- Java并发编程的问题
- Actor并发编程模型
- Java并发编程对比Actor并发编程
- 创建Actor
- Actor程序运行流程
- 发送消息/接收消息
- 持续接收消息
- 使用loop和react优化接收消息
- 发送和接收自定义消息

Actor介绍

scala的Actor并发编程模型可以用来开发比Java线程效率更高的并发程序。我们学习scala Actor的目的主要是为后续学习Akka做准备。

Java并发编程的问题

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

线程死锁

Actor并发编程模型

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

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

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

scala在2.11.x版本中加入了Akka并发编程框架，老版本已经废弃。Actor的编程模型和Akka很像，我们这里学习Actor的目的是为学习Akka做准备。

创建Actor

创建Actor的方式和Java中创建线程很类似，也是通过继承来创建。

使用方式

1.定义class或object继承Actor特质
2.重写act方法
3.调用Actor的start方法执行Actor

类似于Java线程，这里的每个Actor是并行执行的

示例

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

使用class继承Actor创建（如果需要在程序中创建多个相同的Actor）
使用object继承Actor创建（如果在程序中只创建一个Actor）

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

object _05ActorDemo {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 =for(i <- 1 to 10) {println(i)}
}
object Actor2 extends Actor {override def act(): Unit =for(i <- 11 to 20) {println(i)}
}
def main(args: Array[String]): Unit = {Actor1.start()Actor2.start()
}

Actor程序运行流程

1.调用start()方法启动Actor
2.自动执行act()方法
3.向Actor发送消息
4.act方法执行完成后，程序会调用exit()方法

发送消息/接收消息

    我们之前介绍Actor的时候，说过Actor是基于事件（消息）的并发编程模型，那么Actor是如何发送消息和接收消息的呢？

使用方式
我们可以使用三种方法来发送消息：

!	发送异步消息，没用返回值
！？	发送同步消息，等待返回值
！！	发送异步消息，返回值是Future[Any]

例如:
要给actor1发送一个异步字符串消息，使用以下代码：

actor1 ! “你好!”

接收消息

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

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

receive方法只接收一次消息，接收完后继续执行act方法
示例
创建两个Actor（ActorSender、ActorReceiver）
ActorSender发送一个异步字符串消息给ActorReceiver
ActorReceive接收到该消息后，打印出来

参考代码

  object ActorSender extends Actor {override def act(): Unit = {// 发送消息while(true) {ActorReceiver ! "hello!"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 = {ActorReceiver.start()ActorSender.start()}

持续接收消息

    通过上一个案例，ActorReceiver调用receive来接收消息，但接收一次后，Actor就退出了。

object ActorSender extends Actor {override def act(): Unit = {// 发送字符串消息给Actor2val 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来接收消息就可以啦。

示例

在上一个案例的基础上，让ActorReceiver能够一直接收消息

  object ActorSender extends Actor {override def act(): Unit = {// 发送消息while(true) {ActorReceiver ! "hello!"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 = {ActorReceiver.start()ActorSender.start()}

使用loop和react优化接收消息

上述代码，使用while循环来不断接收消息。

如果当前Actor没有接收到消息，线程就会处于阻塞状态
如果有很多的Actor，就有可能会导致很多线程都是处于阻塞状态
每次有新的消息来时，重新创建线程来处理

频繁的线程创建、销毁和切换，会影响运行效率

  在scala中，可以使用loop + react来复用线程。比while + receive更高效

示例

使用loop + react重写上述案例

参考代码

// 持续接收消息
loop {react {case msg:String => println("接收到消息：" + msg)}
}

发送和接收自定义消息

    我们前面发送的消息是字符串类型，Actor中也支持发送自定义消息，常见的如：使用样例类封装消息，然后进行发送处理。

示例一

创建一个MsgActor，并向它发送一个同步消息，该消息包含两个字段（id、message）
MsgActor回复一个消息，该消息包含两个字段（message、name）
打印回复消息
使用!?来发送同步消息
在Actor的act方法中，可以使用sender获取发送者的Actor引用

  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）
使用!发送异步无返回消息

case class Mesasge(message:String, company:String)object MsgActor extends Actor {override def act(): Unit = {loop {react {case Mesasge(message, company) =>println(s"MsgActor接收到消息:${message}/${company}")}}}
}def main(args: Array[String]): Unit = {MsgActor.start()MsgActor ! Mesasge("中国联通", "大爷，快交话费！")
}

示例三

创建一个MsgActor，并向它发送一个异步有返回消息，该消息包含两个字段（id、message）
MsgActor回复一个消息，该消息包含两个字段（message、name）
打印回复消息
使用!!发送异步有返回消息
发送后，返回类型为Future[Any]的对象
Future表示异步返回数据的封装，虽获取到Future的返回值，但不一定有值，可能在将来某一时刻才会返回消息
Future的isSet()可检查是否已经收到返回消息，apply()方法可获取返回数据

参考代码

case class Message(id:Int, message:String)
case class ReplyMessage(message:String, name:String)object MsgActor extends Actor {override def act(): Unit = {loop {react {case Message(id, message) =>println(s"MsgActor接收到消息：${id}/${message}")sender ! ReplyMessage("收到消息！", "JIm")}}}
}def main(args: Array[String]): Unit = {MsgActor.start()val future: Future[Any] = MsgActor !! Message(1, "你好！")while(!future.isSet) {}val replyMessage = future.apply().asInstanceOf[ReplyMessage]println(replyMessage)
}

ScalaActor介绍