2. 程式人生 > >Akka併發程式設計——第三節：Actor模型（二）

Akka併發程式設計——第三節：Actor模型（二）

阿新 • • 發佈：2018-12-25

本節主要內容：

  1. Actor API解析

1. Actor API解析

Actor中的主要成員變數和方法定義如下：

package akka.actor
trait Actor extends scala.AnyRef {
  type Receive = akka.actor.Actor.Receive

  //context變數暴露當前Actor的上下文資訊及當前訊息
  implicit val context : akka.actor.ActorContext = { /* compiled code */ }

  //self作為當前ActorRef的引用
  implicit final val self : akka.actor.ActorRef = { /* compiled code */ }

  //當前Actor接收到最後一條訊息對應的訊息傳送者(Actor）
  final def
 
 sender() : akka.actor.ActorRef = { /* compiled code */ }

  //receive方法，抽象方法，定義Actor的行為邏輯
  def receive : akka.actor.Actor.Receive

  //內部使用API 
  protected[akka] def aroundReceive(receive : akka.actor.Actor.Receive, msg : scala.Any) : scala.Unit = { /* compiled code */ }
  protected[akka] def aroundPreStart
 
() : scala.Unit = { /* compiled code */ }
  protected[akka] def aroundPostStop() : scala.Unit = { /* compiled code */ }
  protected[akka] def aroundPreRestart(reason : scala.Throwable, message : scala.Option[scala.Any]) : scala.Unit = { /* compiled code */ }
  protected[akka] def aroundPostRestart(reason : scala.Throwable)
 
 : scala.Unit = { /* compiled code */ }


   //監督策略，用於Actor容錯處理
  def supervisorStrategy : akka.actor.SupervisorStrategy = { /* compiled code */ }

  //Hook方法，用於Actor生命週期監控 
  @scala.throws[T](classOf[scala.Exception])
  def preStart() : scala.Unit = { /* compiled code */ }
  @scala.throws[T](classOf[scala.Exception])
  def postStop() : scala.Unit = { /* compiled code */ }
  @scala.throws[T](classOf[scala.Exception])
  def preRestart(reason : scala.Throwable, message : scala.Option[scala.Any]) : scala.Unit = { /* compiled code */ }
  @scala.throws[T](classOf[scala.Exception])
  def postRestart(reason : scala.Throwable) : scala.Unit = { /* compiled code */ }


  //傳送給Actor的訊息，Actor沒有定義相應的處理邏輯時，會呼叫此方法
  def unhandled(message : scala.Any) : scala.Unit = { /* compiled code */ }
}
object Actor extends scala.AnyRef {
  type Receive = scala.PartialFunction[scala.Any, scala.Unit]

  //空的行為邏輯
  @scala.SerialVersionUID(1)
  object emptyBehavior extends scala.AnyRef with akka.actor.Actor.Receive {
    def isDefinedAt(x : scala.Any) : scala.Boolean = { /* compiled code */ }
    def apply(x : scala.Any) : scala.Nothing = { /* compiled code */ }
  }
  //Sender為null
  @scala.SerialVersionUID(1)
  final val noSender : akka.actor.ActorRef = { /* compiled code */ }
}

（1） Hook方法，preStart()、postStop()方法的使用

/*
 *Actor API: Hook方法
 */
  object Example_05 extends App{
    import akka.actor.Actor
    import akka.actor.ActorSystem
    import akka.actor.Props


    class FirstActor extends Actor with ActorLogging{
      //通過context.actorOf方法建立Actor
      var child:ActorRef = _

      //Hook方法，preStart()，Actor啟動之前呼叫，用於完成初始化工作
      override def preStart(): Unit ={
        log.info("preStart() in FirstActor")
        //通過context上下文建立Actor
        child = context.actorOf(Props[MyActor], name = "myChild")
      }
      def receive = {
        //向MyActor傳送訊息
        case x => child ! x;log.info("received "+x)
      }

      //Hook方法，postStop()，Actor停止之後呼叫
      override def postStop(): Unit = {
        log.info("postStop() in FirstActor")
       }
    }


    class MyActor extends Actor with ActorLogging{
      //Hook方法，preStart()，Actor啟動之前呼叫，用於完成初始化工作
      override def preStart(): Unit ={
        log.info("preStart() in MyActor")
      }
      def receive = {
        case "test" => log.info("received test")
        case _      => log.info("received unknown message")
      }

      //Hook方法，postStop()，Actor停止之後呼叫
      override def postStop(): Unit = {
        log.info("postStop() in MyActor")
      }
    }

    val system = ActorSystem("MyActorSystem")
    val systemLog=system.log

    //建立FirstActor物件
    val myactor = system.actorOf(Props[FirstActor], name = "firstActor")

    systemLog.info("準備向myactor傳送訊息")
    //向myactor傳送訊息
    myactor!"test"
    myactor! 123
    Thread.sleep(5000)
    //關閉ActorSystem，停止程式的執行
    system.shutdown()
  }

程式碼執行結果：

[INFO] [04/02/2016 17:04:49.607] [main] [ActorSystem(MyActorSystem)] 準備向myactor傳送訊息
[INFO] [04/02/2016 17:04:49.607] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor] preStart() in FirstActor
[INFO] [04/02/2016 17:04:49.607] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor] received test
[INFO] [04/02/2016 17:04:49.607] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor] received 123
[INFO] [04/02/2016 17:04:49.608] [MyActorSystem-akka.actor.default-dispatcher-2] [akka://MyActorSystem/user/firstActor/myChild] preStart() in MyActor
[INFO] [04/02/2016 17:04:49.608] [MyActorSystem-akka.actor.default-dispatcher-2] [akka://MyActorSystem/user/firstActor/myChild] received test
[INFO] [04/02/2016 17:04:49.608] [MyActorSystem-akka.actor.default-dispatcher-2] [akka://MyActorSystem/user/firstActor/myChild] received unknown message
[INFO] [04/02/2016 17:04:54.616] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor/myChild] postStop() in MyActor
[INFO] [04/02/2016 17:04:54.617] [MyActorSystem-akka.actor.default-dispatcher-3] [akka://MyActorSystem/user/firstActor] postStop() in FirstActor

在程式碼

 class FirstActor extends Actor with ActorLogging{
      //通過context.actorOf方法建立Actor
      var child:ActorRef = _

      //Hook方法，preStart()，Actor啟動之前呼叫，用於完成初始化工作
      override def preStart(): Unit ={
        log.info("preStart() in FirstActor")
        //通過context上下文建立Actor
        child = context.actorOf(Props[MyActor], name = "myChild")
      }
      def receive = {
        //向MyActor傳送訊息
        case x => child ! x;log.info("received "+x)
      }

      //Hook方法，postStop()，Actor停止之後呼叫，用於完成初始化工作
      override def postStop(): Unit = {
        log.info("postStop() in FirstActor")
      }
    }

中分別對postStop、preStart方法進行了重寫，在preStart方法中通過程式碼

 child = context.actorOf(Props[MyActor], name = "myChild")

對成員變數child進行初始化，然後在postStop方法中使用

 //通過context上下文停止MyActor的執行
        context.stop(child)

停止MyActor的執行。在使用程式碼

//建立FirstActor物件
val myactor = system.actorOf(Props[FirstActor], name = "firstActor")

建立FirstActor時，便會呼叫preStart方法完成MyActor的建立，因此首先會執行FirstActor中的preStart()方法

dispatcher-4] [akka://MyActorSystem/user/firstActor] preStart() in FirstActor

然後在建立MyActor時執行MyActor中定義的preStart方法

[INFO] [04/02/2016 17:04:49.608] [MyActorSystem-akka.actor.default-dispatcher-2] [akka://MyActorSystem/user/firstActor/myChild] preStart() in MyActor

在執行

//關閉ActorSystem，停止程式的執行
    system.shutdown()

FirstActor作為MyActor的Supervisor，會先停止MyActor，再停止自身，因此先呼叫MyActor的postStop方法，再呼叫FirstActor的postStop方法。

（2） 成員變數self及成員方法sender方法的使用

整體程式碼如下：

  /*
 *Actor API:成員變數self及sender()方法的使用
 */
  object Example_05 extends App{
    import akka.actor.Actor
    import akka.actor.ActorSystem
    import akka.actor.Props


    class FirstActor extends Actor with ActorLogging{
      //通過context.actorOf方法建立Actor
      var child:ActorRef = _

      override def preStart(): Unit ={
        log.info("preStart() in FirstActor")
        //通過context上下文建立Actor
        child = context.actorOf(Props[MyActor], name = "myActor")
      }
      def receive = {
        //向MyActor傳送訊息
        case x => child ! x;log.info("received "+x)
      }


    }


    class MyActor extends Actor with ActorLogging{
      self!"message from self reference"
      def receive = {
        case "test" => log.info("received test");sender()!"message from MyActor"
        case "message from self reference"=>log.info("message from self refrence")
        case _      => log.info("received unknown message");
      }

    }

    val system = ActorSystem("MyActorSystem")
    val systemLog=system.log

    //建立FirstActor物件
    val myactor = system.actorOf(Props[FirstActor], name = "firstActor")

    systemLog.info("準備向myactor傳送訊息")
    //向myactor傳送訊息
    myactor!"test"
    myactor! 123
    Thread.sleep(5000)
    //關閉ActorSystem，停止程式的執行
    system.shutdown()
  }

執行結果：

[INFO] [04/02/2016 18:40:37.805] [main] [ActorSystem(MyActorSystem)] 準備向myactor傳送訊息
[INFO] [04/02/2016 18:40:37.805] [MyActorSystem-akka.actor.default-dispatcher-3] [akka://MyActorSystem/user/firstActor] preStart() in FirstActor
[INFO] [04/02/2016 18:40:37.806] [MyActorSystem-akka.actor.default-dispatcher-3] [akka://MyActorSystem/user/firstActor] received test
[INFO] [04/02/2016 18:40:37.806] [MyActorSystem-akka.actor.default-dispatcher-3] [akka://MyActorSystem/user/firstActor] received 123
[INFO] [04/02/2016 18:40:37.806] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor/myActor] received test
[INFO] [04/02/2016 18:40:37.806] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor/myActor] message from self refrence
[INFO] [04/02/2016 18:40:37.806] [MyActorSystem-akka.actor.default-dispatcher-3] [akka://MyActorSystem/user/firstActor] received message from MyActor
[INFO] [04/02/2016 18:40:37.806] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor/myActor] received unknown message
[INFO] [04/02/2016 18:40:37.806] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor/myActor] received unknown message

程式碼：

class MyActor extends Actor with ActorLogging{
      self!"message from self reference"
      def receive = {
        case "test" => log.info("received test");sender()!"message from MyActor"
        case "message from self reference"=>log.info("message from self refrence")
        case _      => log.info("received unknown message");
      }

    }

中使用

self!"message from self reference"

向自身傳送了一條訊息，receive方法通過

        case "message from self reference"=>log.info("message from self refrence")

對這條訊息進行處理。receive方法在處理

def receive = {
        case "test" => log.info("received test");sender()!"message from MyActor"

“test”訊息時，會呼叫

sender()!"message from MyActor"

向sender（本例中為FirstActor）傳送”message from MyActor”訊息，FirstActor使用

 def receive = {
        //向MyActor傳送訊息
        case x => child ! x;log.info("received "+x)
      }

處理訊息時又向MyActor回送該訊息，因此最終的輸出有兩個unknown message，分別對應123和”message from MyActor”

[INFO] [04/02/2016 18:40:37.806] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor/myActor] received unknown message
[INFO] [04/02/2016 18:40:37.806] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor/myActor] received unknown message

（3） unhandled方法的使用

unhandled方法用於處理沒有被receive方法處理的訊息，下面的程式碼給出的是當不重寫unhandled方法時的程式碼

/*
*Actor API:unhandled方法
*/
object Example_06 extends App{
  import akka.actor.Actor
  import akka.actor.ActorSystem
  import akka.actor.Props
  class FirstActor extends Actor with ActorLogging{
    def receive = {
      //向MyActor傳送訊息
      case "test" => log.info("received test")
    }


  }
  val system = ActorSystem("MyActorSystem")
  val systemLog=system.log

  //建立FirstActor物件
  val myactor = system.actorOf(Props[FirstActor], name = "firstActor")

  systemLog.info("準備向myactor傳送訊息")
  //向myactor傳送訊息
  myactor!"test"
  myactor! 123
  Thread.sleep(5000)
  //關閉ActorSystem，停止程式的執行
  system.shutdown()
}

程式碼輸出：

[INFO] [04/02/2016 19:14:11.992] [main] [ActorSystem(MyActorSystem)] 準備向myactor傳送訊息
[INFO] [04/02/2016 19:14:11.992] [MyActorSystem-akka.actor.default-dispatcher-3] [akka://MyActorSystem/user/firstActor] received test

不難看出，對於

 myactor! 123

傳送的這條訊息沒有被處理，沒有任何的處理邏輯。在實際開發過程中，可能會對不能被處理的訊息增加一些應對邏輯，此時可以重寫unhandled方法，程式碼如下：

/*
*Actor API:unhandled方法的使用
*/
object Example_06 extends App{
  import akka.actor.Actor
  import akka.actor.ActorSystem
  import akka.actor.Props


  class FirstActor extends Actor with ActorLogging{
    def receive = {
      //向MyActor傳送訊息
      case "test" => log.info("received test")
    }

    //重寫unhandled方法
    override def unhandled(message: Any): Unit = {
      log.info("unhandled message is {}",message)
    }
  }



  val system = ActorSystem("MyActorSystem")
  val systemLog=system.log

  //建立FirstActor物件
  val myactor = system.actorOf(Props[FirstActor], name = "firstActor")

  systemLog.info("準備向myactor傳送訊息")
  //向myactor傳送訊息
  myactor!"test"
  myactor! 123
  Thread.sleep(5000)
  //關閉ActorSystem，停止程式的執行
  system.shutdown()
}

程式碼輸出結果：

[INFO] [04/02/2016 19:17:18.458] [main] [ActorSystem(MyActorSystem)] 準備向myactor傳送訊息
[INFO] [04/02/2016 19:17:18.458] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor] received test
[INFO] [04/02/2016 19:17:18.458] [MyActorSystem-akka.actor.default-dispatcher-4] [akka://MyActorSystem/user/firstActor] unhandled message is 123

其它如preRestart等方法的使用將在Akka容錯部分進行講解。

Scala學習（公眾微訊號：ScalaLearning）每天為大家帶來一點Scala語言、Spark、Kafka、Flink、AKKA等大資料技術乾貨及相關技術資訊。技術永無止境，勇攀高峰，一往直前！
覺得文章不錯？掃描關注
這裡寫圖片描述

相關推薦

Akka併發程式設計——第三節：Actor模型（二）

本節主要內容： Actor API解析 1. Actor API解析 Actor中的主要成員變數和方法定義如下： package akka.actor trait Actor extends scala.AnyRef { type Rec

Akka併發程式設計——第八節：Actor模型（七）

本節主要內容 停止執行Typed Actor 當Typed Actor不再需要時要將其停止，有3種方法停止Typed Actor的執行： （1）通過system.shutdown()停止ActorSystem中所有的Typed Actor；

Akka併發程式設計——第二節：Actor模型(一）

本節主要內容 定義Actor 建立Actor 1. 定義Actor 通過擴充套件akka.actor.Actor 特質並實現receive方法來定義Actor，程式碼示例如下 //通過擴充套件Actor並實現receive方法來定義Actor

Akka併發程式設計——第七節：Actor模型（六）

主要內容： 1. Typed Actor定義 2. Typed Actor建立 3. 訊息傳送 1. Typed Actor定義 Akka中的Typed Actor是Active Objects設計模式的實現，Active Objects模式將方法的執

Akka併發程式設計——第六節：Actor模型（五）

本將主要內容： 1. !訊息傳送，Fire-and-Forget訊息模型 2. ？訊息傳送，Send-And-Receive-Future訊息模型 Akka提供了兩種訊息模型：fire-and-forget和Send-And-Receive-Future。

Akka併發程式設計——第五節：Actor模型（四）

本節主要內容： 1. 停止Actor 1. 停止Actor （1）通過ActorSystem.shutdown方法停止所有 Actor的執行 /* *停止Actor:ActorSystem.shutdown方法 */ object Example_1

Akka併發程式設計——第四節：Actor模型（三）

本將主要內容： 1. Actor引用、Actor路徑 1. Actor引用、Actor路徑 下圖是Akka官方文件中給出的一張圖 該圖清晰地說明了ActorPath，ActorRef，Actor及ActorSystem之間的關係，並說明了Actor整

F28027第三節課---系統時鐘（二）

今天還是繼續學習時鐘 今天看晶片手冊的時候，發現自己昨天好敷衍。。今天必須要來做點補充，不然虛的慌 第一、安全裝置watchdog 首先看下watchdog整體裝置： 看門狗要復位輸出脈衝訊號，從Generate Output Pulsemo模組來看，有兩個控制條件：SC

scala之Akka的Actor模型（上）

原文地址：http://my.oschina.net/jingxing05/blog/287213 明確並行和併發 看兩張圖 並行parallelism 併發concurrency 關鍵點在於 多個任務之間 有沒有 交流和互動的！ 交流： 共享記憶體

線性分類模型（二）：logistic迴歸模型分析

前言 上一篇文章介紹了線性判別模型，本文介紹線性生成模型——logistic迴歸模型。本文介紹logstic迴歸模型相關的知識，為了更好理解模型的決策邊界函式，本文同時分析了多元變數的協方差對概率分佈的影響。   目錄 1、logistic迴歸模型的含義 2、l

併發模型（二）——Master-Worker模式

    Master-Worker模式是常用的並行模式之一，它的核心思想是，系統有兩個程序協作工作：Master程序，負責接收和分配任務；Worker程序，負責處理子任務。當Worker程序將子任務處理完成後，結果返回給Master程序，由Master程序做歸納彙總，最後得到最

windows 程式設計 第三章讀書筆記（上）

     孃的，今天晚上在陽臺做飯把水管一腳踢爆了，水流到下面的住戶的陽臺，讓個老孃們把我罵了一頓，本著做錯事的原則，我裝的很紳士還說了個對不起，擦，真是條純漢子，能屈能伸。    

JVM——記憶體模型（二）：虛擬機器棧與本地方法棧

本篇文章將繼續認識Java虛擬機器中的記憶體模型，今天要認識的是我們常說的"棧”。 棧其實也分兩種，一種是虛擬機器棧，一種是本地方法棧。而我們平常說的最多的，就是虛擬機器棧。接下來就讓我們走進這兩個棧，看看他們是個啥。 1.虛擬機器棧 虛擬機器棧，即Java Virtual&n

函式程式設計實驗四：列表練習（2）

{- 吳坎 17341163 [email protected] 資料科學與計算機學院計算機專業 -} {- 稱一個三元組(x,y,z)是畢達哥拉斯三元組，如果x*x + y*y == z*

C語言程式設計第八章8.1（3）

** 利用矩陣相乘公式程式設計計算mn階矩陣A和nm階矩陣B之積 ** #include<stdio.h> #define ROW 2 #define COL 3 /*函式功能：計算矩陣相乘之積，結果存於二維陣列c中*/ void MultiplyM

python3程式設計基礎：多程序（二）建立程序

在之前的文章中對多程序的一些基礎概念，程序的生命週期和python程序操作的模組做了說明，本篇文章直接上程式碼，結束python中建立多程序的一些方法。 os.fork()（Linux） fork()函式，只在Linux系統下存在。而且它非常特殊，普通的函式呼叫，呼叫一次，返回一次，

程式設計菜鳥到大佬之路：資料結構（二）

資料結構C++版——鄧俊輝課堂筆記 第一章 緒論 複雜度度量 時間複雜度 問題例項的規模往往是決定計算成本的主要因素。一般地，問題規模越接近，相應的計算成本也越接近；而隨著問題規模的擴大，計算成本通常也呈上升趨勢。 執行時間的這一變化

程式設計菜鳥到大佬之路：演算法基礎（二）

第二天學習精要 列舉 生理週期 例題2：生理週期 題目描述 人有體力、情商、智商的高峰日子，它們分別每隔23天、 28天和33天出現一次。 對於每個人，我們想知道何時三個高峰落在同一天。 給

Java併發程式設計——執行緒池的使用（五）延時執行的執行緒池ScheduledExecutorService

一、ScheduledExecutorService的延時執行功能 ScheduledExecutorService是延時執行的執行緒池，推薦用ScheduledExecutorService代替timer定時器。 建立一個ScheduledExecutor

Java併發程式設計——執行緒池的使用（三）執行緒池執行任務、取消任務

一、執行緒池執行Runnable任務 executor.execute(runnable) executor.execute(new Runnable() { @Override public void run(