AKKA学习笔记
2017-02-10 20:03
357 查看
AKKA学习笔记总结
Spark 的RPC是通过Akka类库实现的,Akka用Scala语言开发,基于Actor并发模型实现,Akka具有高可靠、高性能、可扩展、分布式等特点,使用Akka可以轻松实现分布式RPC功能。
Actor模型:在计算机科学领域,Actor模型是一个并行计算模型,它把actor作为并行计算的基本元素来对待:为响应一个接收到的消息,一个actor能够自己做出一些决策,如创建更多的actor,或发送更多的消息,或者确定如何去响应接收到的下一个消息。
2. Scala中的多线程
(1) 传统的并发是通过线程(thread)来实现的。在传统的并发模型中,程序被分成若干份同时执行的任务,并且所有任务都对一块共享的内存进行操作。在传统的并发模型会引起竞争问题,可以采取锁机制避免竞争问题,但同时这可能带来死锁等问题。
(2) 在Scala中,多线程的基础就是Actor,核心思想是用消息传递来进行线程间的信息共享和同步。它是基于事件模型的并发机制,Scala是运用消息(message)的发送、接收来实现多线程的。使用Scala能够更容易地实现多线程应用的开发。
Actor模型是另一种不同的并发模型,它很好地解决了在传统并发模型中竞争和死锁等问题。我们可以把一个由actor模型实现的并发程序看成是一个星系一样,星系里面有很多星球,每个星球都是一个actor,星球之间不共享任何资源,但是它们之间有通道来相互传递信息。
每个星球(actor)都有一个信箱来接受来自其它星球的任意信息,它会按照信息接收的顺序来处理,处理完一个信息然后接着处理下一个信息。可以按照信息类型来触发不同的行为。
同时,每个星球(actor)可以异步地(也可以同步,但不是这里谈论的重点)向其它任意星球发送任意消息,就是说,它发送消息之后不会等待返回信息而是直接执行接下来的操作。
比如:
启动线程:
关于Actor之间的消息传递不是详情见之后的一篇学习笔记。
1. 架构图
2. 重要类和方法
ActorSystem
在Akka中,ActorSystem是一个重量级的结构,他按需分配多个线程,所以在实际应用中,ActorSystem通常是一个单例对象,我们可以使用这个ActorSystem创建很多Actor。
在Akka中,Actor负责通信,在Actor中有一些重要的生命周期方法。
1.preStart()方法:该方法在Actor对象构造方法执行后执行,整个Actor生命周期中仅执行一次。
2.receive()方法:该方法在Actor的preStart方法执行完成后执行,用于接收消息,会被反复执行。
3. 实现
Master类
Worker类
暂时先记录到这儿吧,再完善。
参考:《http://blog.csdn.net/fancylovejava/article/details/24724395》
01. AKKA
1. 介绍:
Akka基于Actor模型,提供了一个用于构建可扩展的(Scalable)、弹性的(Resilient)、快速响应的(Responsive)应用程序的平台。2. Spark中的RPC
目前大多数的分布式架构底层通信都是通过RPC(进程间通信)实现的,比如Hadoop项目的RPC通信框架,但是Hadoop在设计之初就是为了运行长达数小时的批量而设计的,在某些极端的情况下,任务提交的延迟很高,所有Hadoop的RPC显得有些笨重。Spark 的RPC是通过Akka类库实现的,Akka用Scala语言开发,基于Actor并发模型实现,Akka具有高可靠、高性能、可扩展、分布式等特点,使用Akka可以轻松实现分布式RPC功能。
3. Actor模型
1. 介绍Actor模型:在计算机科学领域,Actor模型是一个并行计算模型,它把actor作为并行计算的基本元素来对待:为响应一个接收到的消息,一个actor能够自己做出一些决策,如创建更多的actor,或发送更多的消息,或者确定如何去响应接收到的下一个消息。
2. Scala中的多线程
(1) 传统的并发是通过线程(thread)来实现的。在传统的并发模型中,程序被分成若干份同时执行的任务,并且所有任务都对一块共享的内存进行操作。在传统的并发模型会引起竞争问题,可以采取锁机制避免竞争问题,但同时这可能带来死锁等问题。
(2) 在Scala中,多线程的基础就是Actor,核心思想是用消息传递来进行线程间的信息共享和同步。它是基于事件模型的并发机制,Scala是运用消息(message)的发送、接收来实现多线程的。使用Scala能够更容易地实现多线程应用的开发。
Actor模型是另一种不同的并发模型,它很好地解决了在传统并发模型中竞争和死锁等问题。我们可以把一个由actor模型实现的并发程序看成是一个星系一样,星系里面有很多星球,每个星球都是一个actor,星球之间不共享任何资源,但是它们之间有通道来相互传递信息。
每个星球(actor)都有一个信箱来接受来自其它星球的任意信息,它会按照信息接收的顺序来处理,处理完一个信息然后接着处理下一个信息。可以按照信息类型来触发不同的行为。
同时,每个星球(actor)可以异步地(也可以同步,但不是这里谈论的重点)向其它任意星球发送任意消息,就是说,它发送消息之后不会等待返回信息而是直接执行接下来的操作。
比如:
object MyActor1 extends Actor{ //重新act方法 def act(){ for(i <- 1 to 20){ println("actor-1 " + i) Thread.sleep(1000) } } }
启动线程:
//启动Actor MyActor1.start()
关于Actor之间的消息传递不是详情见之后的一篇学习笔记。
4. AKKA实现RPC
Scala在2.11.x版本中将Akka加入其中,作为其默认的Actor,老版本的Actor已经废弃。1. 架构图
2. 重要类和方法
ActorSystem
在Akka中,ActorSystem是一个重量级的结构,他按需分配多个线程,所以在实际应用中,ActorSystem通常是一个单例对象,我们可以使用这个ActorSystem创建很多Actor。
在Akka中,Actor负责通信,在Actor中有一些重要的生命周期方法。
1.preStart()方法:该方法在Actor对象构造方法执行后执行,整个Actor生命周期中仅执行一次。
2.receive()方法:该方法在Actor的preStart方法执行完成后执行,用于接收消息,会被反复执行。
3. 实现
Master类
package wrd.akka import akka.actor.Actor import akka.actor.ActorSystem import com.typesafe.config.ConfigFactory import akka.actor.Props class Master_old extends Actor { println("constructor invoked") override def preStart(): Unit = { println("preStart invoked") } //用于接收消息,sender就是发送者的代理 def receive: Actor.Receive = { case "connect" => { println("a client connected") sender ! "reply" } case "hello" => { println("hello") } } } object Master_old { def main(args: Array[String]): Unit = { val host = "127.0.0.1" val port = 8888 // 准备配置 val configStr = s""" |akka.actor.provider = "akka.remote.RemoteActorRefProvider" |akka.remote.netty.tcp.hostname = "$host" |akka.remote.netty.tcp.port = "$port" """.stripMargin val config = ConfigFactory.parseString(configStr) //ActorSystem老大,辅助创建和监控下面的Actor,他是单例的 val actorSystem = ActorSystem("MasterSystem", config) val master = actorSystem.actorOf(Props(new Master_old), "Master") //Master主构造器会执行 master ! "hello" //发送信息 actorSystem.awaitTermination() //让进程等待着, 先别结束 } }
Worker类
package wrd.akka import akka.actor.Actor import akka.actor.ActorSelection import akka.actor.ActorSystem import com.typesafe.config.ConfigFactory import akka.actor.Props class Worker_old(val masterHost: String, val masterPort: Int) extends Actor { var master: ActorSelection = _ //建立连接 override def preStart(): Unit = { //在master启动时会打印下面的那个协议, 可以先用这个做一个标志, 连接哪个master //继承actor后会有一个context, 可以通过它来连接 master = context.actorSelection(s"akka.tcp://MasterSystem@$masterHost:$masterPort/user/Master") //需要有/user, Master要和master那边创建的名字保持一致 master ! "connect" } def receive: Actor.Receive = { case "reply" => { println("a reply from master") } } } object Worker_old { def main(args: Array[String]): Unit = { val host = "127.0.0.1" val port = 9999 val masterHost = "127.0.0.1" val masterPort = 8888 // 准备配置 val configStr = s""" |akka.actor.provider = "akka.remote.RemoteActorRefProvider" |akka.remote.netty.tcp.hostname = "$host" |akka.remote.netty.tcp.port = "$port" """.stripMargin val config = ConfigFactory.parseString(configStr) //ActorSystem老大,辅助创建和监控下面的Actor,他是单例的 val actorSystem = ActorSystem("WorkerSystem", config) actorSystem.actorOf(Props(new Worker_old(masterHost, masterPort)), "Worker") actorSystem.awaitTermination() } }
暂时先记录到这儿吧,再完善。
参考:《http://blog.csdn.net/fancylovejava/article/details/24724395》
相关文章推荐
- 详细解释modbus RTU
- 《谈判力》读书笔记:第二章 把人和事分开
- 详析Spring中依赖注入的三种方式
- 自定义View 篇四《低仿QQ测拉删除》
- sass中带中文报错解决方法:
- MFC任务栏中没有名称问题
- 《React Native跨平台移动应用开发》源码
- 腾讯2017 暑假实习生编程题---【3】 有趣的数字
- base stock 订货, (s, S)订货
- Dev支持C++11
- 海思Hi3516A(5)3D降噪
- hibernate,事务
- MFC中OnInitialUpdate问题
- MyEclipse10 破解方法
- JSBridge 从源代码入手<一>
- OkHttp 新手上路
- MFC多个子窗体覆盖
- 《谈判力》读书笔记:第一章 直接在立场上讨价还价的谈判方式的问题
- leetcode:Merge k Sorted Lists (建立堆)
- DevOps企业实践指南(3): 第一条原则:流动