温馨提示×

温馨提示×

您好,登录后才能下订单哦!

密码登录×
登录注册×
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》

Scala的actor

发布时间:2020-07-15 01:02:08 来源:网络 阅读:2070 作者:原生zzy 栏目:开发技术

1. Scala Actor

(1)相关介绍:

  Scala中的actor能够实现并行编程的强大功能,他是基于事件模型的并发机制,scala是运用消息(message)的发送、接收来实现多线程的(Java是使用共享内存实现的)。使用 Scala 能够更容易地实现多线程应用的开发。
  一个actor是一个容器,它包含状态,行为,信箱,子actor和监督策略。所有这些包含在一个actorReference(Actor引用)里。一个actor需要与外界隔离才能从actor模型中获益,所以actor是以actor引用的形式展现给外界的。

(2)Java并发编程模型与scala actor模型的区别:

   1)Java :

     - Java 中的并发编程基本上满足了事件之间相互独立,但是事件不能够同时发生的场景的需要。
     - Java 中的并发编程是基于共享数据和加锁的一种机制,即会有一个共享的数据,然后有 若干个线程去访问这个共享的数据(主要是对这个共享的数据进行修改),同时 Java 利用加锁 的机制(即 synchronized)来确保同一时间只有一个线程对我们的共享数据进行访问,进而保 证共享数据的一致性。
     - Java 中的并发编程存在资源争夺和死锁等多种问题,因此程序越大问题越麻烦。

   2)scala actor :

     - Scala 中的 Actor 是一种不共享数据,依赖于消息传递的一种并发编程模式,避免了死锁、资源争夺等情况。在具体实 现的过程中,Scala 中的 Actor 会不断的循环自己的邮箱,并通过 receive 偏函数进行消息的模式匹配并进行相应的处理。
     - 如果 Actor A 和 Actor B 要相互沟通的话,首先 A 要给 B 传递一个消息,B 会有一个收件箱,然后 B 会不断的循环自己的收件箱,若看见 A 发过来的消息,B 就会解析 A 的消息并执行,处理完之后就有可能将处理的结果通过邮件的方式发送给 A。
Scala的actor
  对于 Java,我们都知道它的多线程实现需要对共享资源(变量、对象等)使用 synchronized 关键字进行代码块同步、对象锁互斥等等。而且,常常一大块的 try…catch 语句块中加上 wait 方法、notify 方法、notifyAll 方法是让人很头疼的。原因就在于 Java 中多数使用的是可变状态的对象资源,对这些资源进行共享来实现多线程编程的话,控制好资源竞争与防止对象状态被意外修改是非常重要的,而对象状态的不变性也是较难以保证的。而在 Scala 中, 我们可以通过复制不可变状态的资源(即对象,Scala 中一切都是对象,连函数、方法也是) 的一个副本,再基于 Actor 的消息发送、接收机制进行并行编程。

(3)actor中发消息的方式:

Scala的actor

(4)actor入门实例:

pom.xml

<properties>
      <scala.version>2.11.8</scala.version>
      <scala.actors.version>2.11.8</scala.actors.version>
</properties>
<dependencies>
      <dependency>
          <groupId>org.scala-lang</groupId>
          <artifactId>scala-library</artifactId>
          <version>${scala.version}</version>
      </dependency>
                  <!-- https://mvnrepository.com/artifact/org.scala-lang/scala-actors -->
      <dependency>
          <groupId>org.scala-lang</groupId>
          <artifactId>scala-actors</artifactId>
          <version>${scala.actors.version}</version>
      </dependency>
</dependencies>
object Base_Actor {
  def main(args: Array[String]): Unit = {
    //调用start方法
    new MyActor1().start()
    new MyActor2().start()
  }
}
////继承Actor类,相当于Java中的Thread
class MyActor1 extends Actor{
  //重新 act 方法,相当于Java中的run方法
  override def act(): Unit = {
    println("hahaha")
  }
}
//继承Actor类,相当于Java中的Thread
class MyActor2 extends Actor{
  //重新 act 方法,相当于Java中的run方法
  override def act(): Unit = {
    println("hello world")
  }
}
// 注意:上面分别调用了两个单例对象的 start()方法,他们的 act()方法会被执行,相同与在 Java 中开启了两个线程,线程的 run()方法会被执行,这两个 Actor 是并行执行的。

2. Akka Actor

  (1) Akka Actor介绍:

  Akka 基于 Actor 模型,提供了一个用于构建可扩展的(Scalable)、弹性的(Resilient)、快速响应的(Responsive)应用程序的平台。
        Scala的actor
  Actor 模型:在计算机科学领域,Actor 模型是一个并行计算(Concurrent Computation)模型, 它把 actor 作为并行计算的基本元素来对待:为响应一个接收到的消息,一个 actor 能够自己做出一些决策,如创建更多的 actor,或发送更多的消息,或者确定如何去响应接收到的下一个消息。
  Actor是Akka中最核心的概念,它是一个封装了状态和行为的对象,Actor之间可以通过交换消息的方式进行通信,每一个actor都有自己的收件箱。通过Actor能够简化锁及线程管理,可以非常容易的开发出正确的并发程序和并行系统。
  Actor 特性: 提供了一种高级抽象,能够简化在并发(Concurrency)/并行(Parallelism)应用场景下 的编程开发 提供了异步非阻塞的、高性能的事件驱动编程模型 超级轻量级事件处理(每 GB 堆内存几百万 Actor)

  (2) Akka Actor 重要API:

  - ActorSystem:在 Akka 中,ActorSystem 是一个重量级的结构,他需要分配多个线程,所以在实际应用中, ActorSystem 通常是一个单例对象,我们可以使用这个 ActorSystem 的 actorOf 方法创建很多 Actor。
  - Actor:在 Akka 中,Actor 负责通信,在 Actor 中有一些重要的生命周期方法。
  - preStart()方法:该方法在 Actor 对象构造方法执行后执行,整个 Actor 生命周期中仅执行一次。
  - receive()方法:该方法在 Actor 的 preStart 方法执行完成后执行,用于接收消息,会被反复执行。

  (3) Akka Actor 入门案例:

      Scala的actor
pom.xml

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <maven.compiler.source>1.7</maven.compiler.source>
    <maven.compiler.target>1.7</maven.compiler.target>
      <scala.version>2.11.8</scala.version>
      <scala.actors.version>2.11.8</scala.actors.version>
      <akka.version>2.4.17</akka.version>
  </properties>
    <dependencies>
      <dependency>
          <groupId>org.scala-lang</groupId>
          <artifactId>scala-library</artifactId>
          <version>${scala.version}</version>
      </dependency>

      <dependency>
          <groupId>com.typesafe.akka</groupId>
          <artifactId>akka-actor_2.11</artifactId>
          <version>${akka.version}</version>
      </dependency>

      <!-- https://mvnrepository.com/artifact/org.scala-lang/scala-actors -->
      <dependency>
          <groupId>org.scala-lang</groupId>
          <artifactId>scala-actors</artifactId>
          <version>${scala.actors.version}</version>
      </dependency>

      <dependency>
          <groupId>com.typesafe.akka</groupId>
          <artifactId>akka-remote_2.11</artifactId>
          <version>${akka.version}</version>
      </dependency>
  </dependencies>

master:

import java.text.SimpleDateFormat
import java.util._

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}

class MyMaster extends  Actor{
  def doHell(): Unit ={
    println("我是master,我接受到了worker的消息!")
  }

  /**
    * 这就是一个偏函数, 用来处理当前这个actor对象接收到的所有的消息
    */
  override def receive: Receive = {
    case "hello" =>{
      doHell
      //用以发送信息到相应的worker,!表示 异步无返回值
      sender() ! "hi"
    }
    case "getNow" =>{
      doHell
      sender() ! new SimpleDateFormat("yyyy-MM-dd").format(new Date())
    }
  }
}
object MyMaster{
  def main(args: Array[String]): Unit = {
    //1.构建一个:ActorSystem
    val strConfig=
      """
        |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
        |akka.remote.netty.tcp.hostname = localhost
        |akka.remote.netty.tcp.port = 9527
      """.stripMargin
    val config: Config = ConfigFactory.parseString(strConfig)
    val myMaster: ActorSystem = ActorSystem("ActorSystemMaster",config)
    //2.通过actorsystem创建actor
    myMaster.actorOf(Props(new MyMaster()),"MasterActor")
  }
}

worker

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.{Config, ConfigFactory}

class MyWorker extends Actor{

  override def preStart(): Unit = {
    val hostname="localhost"
    val serveractorsystem="ActorSystemMaster"
    val serveractor="MasterActor"
    val port="9527"
    //在创建worker actor之前向master发送一个消息
    val master=context.actorSelection(s"akka.tcp://${serveractorsystem}@${hostname}:${port}/user/${serveractor}")
    val message="getNow"
    //获得master相关对象,向master发送信息
    master ! message

  }
  //处理相应的来自master返回的信息
  override def receive: Receive = {
    case date:String => {
      println("时间日期:"+date)
    }
    case "hi" =>{
      println("我是worker,接收到master发送过来的结果: hi")
    }
  }
}
object MyWorker{
  def main(args: Array[String]): Unit = {
    //1.构建一个:ActorSystem
    val strConfig:String=
      """
        |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
        |akka.remote.netty.tcp.hostname = localhost
      """.stripMargin
    val config: Config = ConfigFactory.parseString(strConfig)
    val workerActorSystem: ActorSystem = ActorSystem("workerActorSystem",config)
    workerActorSystem.actorOf(Props(new MyWorker()),"workerActor")
  }
}
向AI问一下细节

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

AI