scala快速上手(九)Scala Actor 通信模型

it2023-05-04  71

Scala Actor 通信模型

 

1. 概念理解

 Java 中的并发编程

Java 中的并发编程是基于共享数据和加锁的一种机制,即会有一个共享的数据,然后有若干个线程去访问这个共享的数据(主要是对这个共享的数据进行修改),同时 Java 利用加锁的机制(即 synchronized)来确保同一时间只有一个线程对我们的共享数据进行访问,进而保证共享数据的一致性。Java 中的并发编程存在资源争夺和死锁等多种问题,因此程序越大问题越麻烦。

 Scala 中的并发编程

Scala 中的并发编程思想与 Java 中的并发编程思想完全不一样,Scala 中的 Actor是一种不共享数据,依赖于消息传递的一种并发编程模式, 避免了死锁、资源争夺等情况。在具体实现的过程中,Scala 中的 Actor 会不断的循环自己的邮箱,并通过 receive 偏函数进行消息的模式匹配并进行相应的处理。

如果 Actor A 和 Actor B 要相互沟通的话,首先 A 要给 B 传递一个消息,B 会有一个收件箱,然后 B 会不断的循环自己的收件箱, 若看见 A 发过来的消息,B就会解析 A 的消息并执行,处理完之后就有可能将处理的结果通过邮件的方式发送给 A。

Actor Model 是用来编写并行计算或分布式系统的高层次抽象(类似 java 中的Thread)让程序员不必为多线程模式下共享锁而烦恼,被用在 Erlang 语言上, 高可用性 99.9999999 % 一年只有 31ms 宕机 Actors 将状态和行为封装在一个轻量的进程/线程中,但是不和其他 Actors 分享状态,每个 Actors 有自己的世界观,当需要和其他 Actors 交互时,通过发送事件和消息,发送是异步的,非堵塞的(fire-andforget),发送消息后不必等另外 Actors 回复,也不必暂停,每个Actors 有自己的消息队列,进来的消息按先来后到排列,这就有很好的并发策略和可伸缩性,可以建立性能很好的事件驱动系统。

Actor 的特征:

 ActorModel 是消息传递模型,基本特征就是消息传递 消息发送是异步的,非阻塞的 消息一旦发送成功,不能修改 Actor 之间传递时,自己决定决定去检查消息,而不是一直等待,是异步非阻塞的

什么是 Akka

Akka 是一个用 Scala 编写的库,用于简化编写容错的、高可伸缩性的 Java 和Scala 的 Actor 模型应用,底层实现就是 Actor,Akka 是一个开发库和运行环境,可以用于构建高并发、分布式、可容错、事件驱动的基于 JVM 的应用。使构建高并发的分布式应用更加容易。

spark1.6 版本之前,spark 分布式节点之间的消息传递使用的就是 Akka,底层也就是 actor 实现的。1.6 之后使用的 netty 传输。

2. 例:Actor 简单例子发送接收消息

import scala.actors.Actor class myActor extends Actor{ def act(){ while(true){ receive { case x:String => println("get String ="+ x) case x:Int => println("get Int") case _ => println("get default") } } } } object Lesson_Actor { def main(args: Array[String]): Unit = { //创建 actor 的消息接收和传递 val actor =new myActor() //启动 actor.start() //发送消息写法 actor ! "i love you !" } }

3. 例:Actor 与 Actor 之间通信

case class Message(actor:Actor,msg:Any) class Actor1 extends Actor{ def act(){ while(true){ receive{ case msg :Message => { println("i sava msg! = "+ msg.msg) msg.actor!"i love you too !" } case msg :String => println(msg) case _ => println("default msg!") } } } } class Actor2(actor :Actor) extends Actor{ actor ! Message(this,"i love you !") def act(){ while(true){ receive{ case msg :String => { if(msg.equals("i love you too !")){ println(msg) actor! "could we have a date !" } } case _ => println("default msg!") } } } } object Lesson_Actor2 { def main(args: Array[String]): Unit = { val actor1 = new Actor1() actor1.start() val actor2 = new Actor2(actor1) actor2.start() } }

 

最新回复(0)