akka actor模型和java thread模型各有啥优势
Posted
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了akka actor模型和java thread模型各有啥优势相关的知识,希望对你有一定的参考价值。
1) akka actor背后也是通过线程池来执行的。actor与thread不是一个维度的事情,actor比thread更抽象,更偏上层,它只是定义了一种机制,怎么实现或底层用什么实现是另一回事。2) 实际上actor更容易驾驭(框架帮你屏蔽了很多底层的事情),thread在并发下兼顾性能和正确性有很多技巧,要用好还没那么容易。 参考技术A 了解java内存模型对深入了解jvm有很多好处。JMM通过控制主内存与每个线程的本地内存之间的交互,来为开发者提供内存可见性保证。
Scala笔记整理:Actor和AKKA
[TOC]
概述
? Scala的Actor有点类似于Java中的多线程编程。但是不同的是,Scala的Actor提供的模型与多线程有所不同。Scala的Actor尽可能地避免锁和共享状态,从而避免多线程并发时出现资源争用的情况,进而提升多线程编程的性能。
Spark中使用的分布式多线程框架,是Akka,是Scala的一种多线程的类库。Akka也实现了类似Scala Actor的模型,其核心概念同样也是Actor。Scala Actor模型已经在2.1.0的时候还在用,但是在2.1.1的时候已经被遗弃了,Spark开始转换用AKKA来替代Scala Actor,但是Scala Actor概念和原理都还是相同的。所以学习Scala Actor对我们学习AKKA,Spark还是有所帮助的
之所以学习Scala Actor,AKKA是因为在学习Spark源码的时,我们能看懂Spark的源码,因为在底层消息传递机制上大量使用AKKA的传送机制。
scala actor
在使用前,需要先引入maven依赖:
<!--scala actor-->
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-actors</artifactId>
<version>2.10.5</version>
</dependency>
actor单向通信
测试代码如下:
package cn.xpleaf.bigdata.p5.myactor
import scala.actors.Actor
/**
* 学习scala actor的基本操作
* 和java中的Runnable Thread几乎一致
*
* 第一步:编写一个类,扩展特质trait Actor(scala 的actor)
* 第二步:复写其中的act方法
* 第三步:创建该actor的对象,调用该对象的start()方法,启动该线程
* 第四步:通过scala的操作符"!",发送消息
* 第五步:结束的话,调用close即可
*
* 模拟单向打招呼
*/
object ActorOps {
def main(args: Array[String]): Unit = {
val mFActor = new MyFirstActor()
mFActor.start()
// 发送消息
mFActor ! "小美,睡了吗?"
mFActor ! "我去洗澡了~"
mFActor ! "呵呵"
}
}
class MyFirstActor extends Actor {
override def act(): Unit = {
while(true) {
receive {
case str: String => println(str)
}
}
}
}
输出结果如下:
小美,睡了吗?
我去洗澡了~
呵呵
使用样例类(case class)进行actor消息传递
测试代码如下:
package cn.xpleaf.bigdata.p5.myactor
import scala.actors.Actor
/**
*
*/
object GreetingActor {
def main(args: Array[String]): Unit = {
val ga = new GreetingActor
ga.start()
ga ! Greeting("小美")
ga ! WorkContent("装系统")
}
}
case class Greeting(name:String)
case class WorkContent(content:String)
class GreetingActor extends Actor {
override def act(): Unit = {
while(true) {
receive {
case Greeting(name) => println(s"Hello, $name")
case WorkContent(content) => println(s"Let‘s talk about sth. with $content")
}
}
}
}
输出结果如下:
Hello, 小美
Let‘s talk about sth. with 装系统
actor相互通信
测试代码如下:
package cn.xpleaf.bigdata.p5.myactor
import scala.actors.Actor
/**
* actor之线程间,互相通信
*
* studentActor
* 向老师问了一个问题
*
* teacherActor
* 向学生做回应
*
* 通信的协议:
* 请求,使用Request(内容)来表示
* 响应,使用Response(内容)来表示
*/
object _03CommunicationActorOps {
def main(args: Array[String]): Unit = {
val teacherActor = new TeacherActor()
teacherActor.start()
val studentActor = new StudentActor(teacherActor)
studentActor.start()
studentActor ! Request("老李啊,scala学习为什么这么难啊")
}
}
case class Request(req:String)
case class Response(resp:String)
class StudentActor(teacherActor: TeacherActor) extends Actor {
override def act(): Unit = {
while(true) {
receive {
case Request(req) => {
// 向老师请求相关的问题
println("学生向老师说:" + req)
teacherActor ! Request(req)
}
case Response(resp) => {
println(resp)
println("高!")
}
}
}
}
}
class TeacherActor() extends Actor {
override def act(): Unit = {
while (true) {
receive {
case Request(req) => { // 接收到学生的请求
sender ! Response("这个问题,需要如此搞定~")
}
}
}
}
}
输出结果如下:
学生向老师说:老李啊,scala学习为什么这么难啊
这个问题,需要如此搞定~
高!
消息的同步和Future
1、Scala在默认情况下,消息都是以异步进行发送的;但是如果发送的消息是同步的,即对方接受后,一定要给自己返回结果,那么可以使用!?的方式发送消息。即:
val response= activeActor !? activeMessage
2、如果要异步发送一个消息,但是在后续要获得消息的返回值,那么可以使用Future。即!!语法,如下:
val futureResponse = activeActor !! activeMessage
val activeReply = future()
AKKA actor
首先需要添加akka的maven依赖:
<!--akka actor-->
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-actor_2.10</artifactId>
<version>2.3.16</version>
</dependency>
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-remote_2.10</artifactId>
<version>2.3.16</version>
</dependency>
<dependency>
<groupId>com.typesafe.akka</groupId>
<artifactId>akka-slf4j_2.10</artifactId>
<version>2.3.16</version>
</dependency>
AKKA消息传递——本地
原理如下:
_01StudentActorOps
package cn.xpleaf.bigdata.p5.myakka.p1
import akka.actor.{Actor, ActorSystem, Props}
import cn.xpleaf.bigdata.p5.myakka.MessageProtocol.{QuoteRequest, QuoteResponse}
import scala.util.Random
/**
* 基于AKKA Actor的单向通信案例
* 学生向老师发送请求
*/
object _01StudentActorOps {
def main(args: Array[String]): Unit = {
// 第一步:构建Actor操作系统
val actorSystem = ActorSystem("StudentActorSystem")
// 第二步:actorSystem创建TeacherActor的代理对象ActorRef
val teacherActorRef = actorSystem.actorOf(Props[TeacherActor])
// 第三步:发送消息
teacherActorRef ! QuoteRequest()
Thread.sleep(2000)
// 第四步:关闭
actorSystem.shutdown()
}
}
class TeacherActor extends Actor {
val quotes = List(
"Moderation is for cowards",
"Anything worth doing is worth overdoing",
"The trouble is you think you have time",
"You never gonna know if you never even try")
override def receive = {
case QuoteRequest() => {
val random = new Random()
val randomIndex = random.nextInt(quotes.size)
val randomQuote = quotes(randomIndex)
val response = QuoteResponse(randomQuote)
println(response)
}
}
}
MessageProtocol
后面akka通信的几个测试程序都会使用到这个object,只在这里给出,后面不再给出。
package cn.xpleaf.bigdata.p5.myakka
/**
* akka actor通信协议
*/
object MessageProtocol {
case class QuoteRequest()
case class QuoteResponse(resp: String)
case class InitSign()
}
object Start extends Serializable
object Stop extends Serializable
trait Message {
val id: String
}
case class Shutdown(waitSecs: Int) extends Serializable
case class Heartbeat(id: String, magic: Int) extends Message with Serializable
case class Header(id: String, len: Int, encrypted: Boolean) extends Message with Serializable
case class Packet(id: String, seq: Long, content: String) extends Message with Serializable
测试
输出结果如下:
QuoteResponse(Anything worth doing is worth overdoing)
AKKA请求与响应——本地
原理如下:
TeacherActor
package cn.xpleaf.bigdata.p5.myakka.p2
import akka.actor.Actor
import cn.xpleaf.bigdata.p5.myakka.MessageProtocol.{QuoteRequest, QuoteResponse}
import scala.util.Random
/**
* Teacher Actor
*/
class TeacherActor extends Actor {
val quotes = List(
"Moderation is for cowards",
"Anything worth doing is worth overdoing",
"The trouble is you think you have time",
"You never gonna know if you never even try")
override def receive = {
case QuoteRequest() => {
val random = new Random()
val randomIndex = random.nextInt(quotes.size)
val randomQuote = quotes(randomIndex)
val response = QuoteResponse(randomQuote)
// println(response)
sender ! response
}
}
}
StudentActor
package cn.xpleaf.bigdata.p5.myakka.p2
import akka.actor.{Actor, ActorLogging, ActorRef}
import cn.xpleaf.bigdata.p5.myakka.MessageProtocol.{InitSign, QuoteRequest, QuoteResponse}
/**
* Student Actor
* 当学生接收到InitSign信号之后,便向老师发送一条Request请求的消息
*/
class StudentActor(teacherActorRef:ActorRef) extends Actor with ActorLogging {
override def receive = {
case InitSign => {
teacherActorRef ! QuoteRequest()
// println("student send request")
}
case QuoteResponse(resp) => {
log.info(s"$resp")
}
}
}
DriverApp
package cn.xpleaf.bigdata.p5.myakka.p2
import akka.actor.{ActorSystem, Props}
import cn.xpleaf.bigdata.p5.myakka.MessageProtocol.InitSign
object DriverApp {
def main(args: Array[String]): Unit = {
val actorSystem = ActorSystem("teacherStudentSystem")
// 老师的代理对象
val teacherActorRef = actorSystem.actorOf(Props[TeacherActor], "teacherActor")
// 学生的代理对象
val studentActorRef = actorSystem.actorOf(Props[StudentActor](new StudentActor(teacherActorRef)), "studentActor")
studentActorRef ! InitSign
Thread.sleep(2000)
actorSystem.shutdown()
}
}
测试
输出结果如下:
[INFO] [04/24/2018 10:02:19.932] [teacherStudentSystem-akka.actor.default-dispatcher-2] [akka://teacherStudentSystem/user/studentActor] Anything worth doing is worth overdoing
AKKA请求与响应——远程
application.conf
MyRemoteServerSideActor {
akka {
actor {
provider = "akka.remote.RemoteActorRefProvider"
}
remote {
enabled-transports = ["akka.remote.netty.tcp"]
netty.tcp {
hostname = "127.0.0.1"
port = 2552
}
}
}
}
MyRemoteClientSideActor {
akka {
actor {
provider = "akka.remote.RemoteActorRefProvider"
}
}
}
RemoteActor
package cn.xpleaf.bigdata.p5.myakka.p3
import akka.actor.{Actor, ActorLogging}
import cn.xpleaf.bigdata.p5.myakka.{Header, Shutdown, Start, Stop}
class RemoteActor extends Actor with ActorLogging {
def receive = {
case Start => { // 处理Start消息
log.info("Remote Server Start ==>RECV Start event : " + Start)
}
case Stop => { // 处理Stop消息
log.info("Remote Server Stop ==>RECV Stop event: " + Stop)
}
case Shutdown(waitSecs) => { // 处理Shutdown消息
log.info("Remote Server Shutdown ==>Wait to shutdown: waitSecs=" + waitSecs)
Thread.sleep(waitSecs)
log.info("Remote Server Shutdown ==>Shutdown this system.")
context.system.shutdown // 停止当前ActorSystem系统
}
case Header(id, len, encrypted) => log.info("Remote Server => RECV header: " + (id, len, encrypted)) // 处理Header消息
case _ =>
}
}
AkkaServerApplication
package cn.xpleaf.bigdata.p5.myakka.p3
import akka.actor.{ActorSystem, Props}
import com.typesafe.config.ConfigFactory
object AkkaServerApplication extends App {
// 创建名称为remote-system的ActorSystem:从配置文件application.conf中获取该Actor的配置内容
val system = ActorSystem("remote-system",
ConfigFactory.load().getConfig("MyRemoteServerSideActor"))
val log = system.log
log.info("===>Remote server actor started: " + system)
// 创建一个名称为remoteActor的Actor,返回一个ActorRef,这里我们不需要使用这个返回值
system.actorOf(Props[RemoteActor], "remoteActor")
}
ClientActor
package cn.xpleaf.bigdata.p5.myakka.p3
import akka.actor.SupervisorStrategy.Stop
import akka.actor.{Actor, ActorLogging}
import cn.xpleaf.bigdata.p5.myakka.{Header, Start}
class ClientActor extends Actor with ActorLogging {
// akka.<protocol>://<actor system>@<hostname>:<port>/<actor path>
val path = "akka.tcp://[email protected]:2552/user/remoteActor" // 远程Actor的路径,通过该路径能够获取到远程Actor的一个引用
val remoteServerRef = context.actorSelection(path) // 获取到远程Actor的一个引用,通过该引用可以向远程Actor发送消息
@volatile var connected = false
@volatile var stop = false
def receive = {
case Start => { // 发送Start消息表示要与远程Actor进行后续业务逻辑处理的通信,可以指示远程Actor初始化一些满足业务处理的操作或数据
send(Start)
if (!connected) {
connected = true
log.info("ClientActor==> Actor connected: " + this)
}
}
case Stop => {
send(Stop)
stop = true
connected = false
log.info("ClientActor=> Stopped")
}
case header: Header => {
log.info("ClientActor=> Header")
send(header)
}
case (seq, result) => log.info("RESULT: seq=" + seq + ", result=" + result) // 用于接收远程Actor处理一个Packet消息的结果
case m => log.info("Unknown message: " + m)
}
private def send(cmd: Serializable): Unit = {
log.info("Send command to server: " + cmd)
try {
remoteServerRef ! cmd // 发送一个消息到远程Actor,消息必须是可序列化的,因为消息对象要经过网络传输
} catch {
case e: Exception => {
connected = false
log.info("Try to connect by sending Start command...")
send(Start)
}
}
}
}
AkkaClientApplication
package cn.xpleaf.bigdata.p5.myakka.p3
import akka.actor.{ActorSystem, Props}
import cn.xpleaf.bigdata.p5.myakka.{Header, Start}
import com.typesafe.config.ConfigFactory
object AkkaClientApplication extends App {
// 通过配置文件application.conf配置创建ActorSystem系统
val system = ActorSystem("client-system",
ConfigFactory.load().getConfig("MyRemoteClientSideActor"))
val log = system.log
val clientActor = system.actorOf(Props[ClientActor], "clientActor") // 获取到ClientActor的一个引用
clientActor ! Start // 发送一个Start消息,第一次与远程Actor握手(通过本地ClientActor进行转发)
Thread.sleep(2000)
clientActor ! Header("What‘s your name: Can you tell me ", 20, encrypted = false) // 发送一个Header消息到远程Actor(通过本地ClientActor进行转发)
Thread.sleep(2000)
}
测试
服务端输出结果如下:
[INFO] [04/24/2018 09:39:49.271] [main] [Remoting] Starting remoting
[INFO] [04/24/2018 09:39:49.508] [main] [Remoting] Remoting started; listening on addresses :[akka.tcp://[email protected]:2552]
[INFO] [04/24/2018 09:39:49.509] [main] [Remoting] Remoting now listens on addresses: [akka.tcp://[email protected]:2552]
[INFO] [04/24/2018 09:39:49.517] [main] [ActorSystem(remote-system)] ===>Remote server actor started: akka://remote-system
[INFO] [04/24/2018 09:46:01.872] [remote-system-akka.actor.default-dispatcher-3] [akka.tcp://[email protected]:2552/user/remoteActor] Remote Server Start ==>RECV Start event : [email protected]
[INFO] [04/24/2018 09:46:03.501] [remote-system-akka.actor.default-dispatcher-3] [akka.tcp://[email protected]:2552/user/remoteActor] Remote Server => RECV header: (What‘s your name: Can you tell me ,20,false)
客户端输出结果如下:
[INFO] [04/24/2018 09:46:01.274] [main] [Remoting] Starting remoting
[INFO] [04/24/2018 09:46:01.479] [main] [Remoting] Remoting started; listening on addresses :[akka.tcp://[email protected]:2552]
[INFO] [04/24/2018 09:46:01.480] [main] [Remoting] Remoting now listens on addresses: [akka.tcp://[email protected]:2552]
[INFO] [04/24/2018 09:46:01.493] [client-system-akka.actor.default-dispatcher-4] [akka.tcp://[email protected]:2552/user/clientActor] Send command to server: [email protected]
[INFO] [04/24/2018 09:46:01.496] [client-system-akka.actor.default-dispatcher-4] [akka.tcp://[email protected]:2552/user/clientActor] ClientActor==> Actor connected: [email protected]
[INFO] [04/24/2018 09:46:03.490] [client-system-akka.actor.default-dispatcher-2] [akka.tcp://[email protected]:2552/user/clientActor] ClientActor=> Header
[INFO] [04/24/2018 09:46:03.491] [client-system-akka.actor.default-dispatcher-2] [akka.tcp://[email protected]:2552/user/clientActor] Send command to server: Header(What‘s your name: Can you tell me ,20,false)
以上是关于akka actor模型和java thread模型各有啥优势的主要内容,如果未能解决你的问题,请参考以下文章