互聯網 - 大數據

上次我們看Actor消息機制，我們看到開火-忘記型消息發出（意思是我們只要發個消息給Actor但是不期望有響應）。

技術上來講， 我們發消息給Actors就是要它的副作用。 這就是這么設計的。除了不響應， 目標Actor當然也可以對消息做以下事情-

1.發送一個響應給發送者（在我們的case里，TeacherActor應該響應一首名言給StudentActor）

2.轉發一個響應給其他的潛在的受眾Actor，他們也可能響應/轉發/產生副作用。Routers（路由）和Supervisors（監管）就是這個例子。（我們馬上就能看到）

REQUEST & RESPONSE
這次，我們只關注第一點- 請求-響應的周期。

圖片說明了我們這次在嘗試歸檔做什么。為了簡介，我沒有在圖中加入ActorSystem, Dispathcer或Mailboxes。

1. DriverApp發送一個InitSignal(初始化信號)消息給StudentActor.

2. StudentActor反饋給InitSignal消息并且發送一條QuoteRequest(格言請求)消息給TeacherActor。

3. TeacherActor，像我們在第一次看到的，響應一條QuoteResponse。

4. StudentActor只記錄QuoteResponse（格言響應）到控制臺/日志記錄器。

我們可以寫個testcase來驗證。

讓我們看下這4點的細節：

1. DRIVERAPP發送一個INITSIGNAL消息給STUDENTACTOR

至此，你可能已經猜出DriverApp會做什么。 就4件事：

1） 初始化ActorSystem

//Initialize the ActorSystem
  val system = ActorSystem("UniversityMessageSystem")

2) 創建TeacherActor

//create the teacher actor
  val teacherRef = system.actorOf(Props[TeacherActor], "teacherActor")

3) 創建StudentActor

//create the Student Actor - pass the teacher actorref as a constructor parameter to StudentActor
  val studentRef = system.actorOf(Props(new StudentActor(teacherRef)), "studentActor")

你可以看到我把ActorRef的實例TeacherActor給StudentActor的構造函數，這樣StudentActor可以使用ActorRef發消息給TeacherActor。還有另外的方式達到這個目的（如傳入Props），但是這個方法對我們后面看到的Supervisors和Routers會方便點。我們也會即將看到child actors但現在還不是一個好方式 - Student創建Teacher聽起來怪怪的吧？

最后，

4） DriverApp會發一個InitSignal給StudentActor, z何陽StudentActor就能開始發送QuoteRequest（格言請求）給TeacherActor。

//send a message to the Student Actor
  studentRef ! InitSignal

這個DriverClass講差不多了。 Thread.sleep和ActorSystem.shutdown在我們最后關閉ActorSystem的時候會在發消息時會等待幾秒保證結束。

** DRIVERAPP.SCALA**

package me.rerun.akkanotes.messaging.requestresponse

import akka.actor.ActorSystem  
import akka.actor.Props  
import me.rerun.akkanotes.messaging.protocols.StudentProtocol._  
import akka.actor.ActorRef

object DriverApp extends App {

  //Initialize the ActorSystem
  val system = ActorSystem("UniversityMessageSystem")

  //construct the teacher actor
  val teacherRef = system.actorOf(Props[TeacherActor], "teacherActor")

  //construct the Student Actor - pass the teacher actorref as a constructor parameter to StudentActor
  val studentRef = system.actorOf(Props(new StudentActor(teacherRef)), "studentActor")

  //send a message to the Student Actor
  studentRef ! InitSignal

  //Let's wait for a couple of seconds before we shut down the system
  Thread.sleep(2000)

  //Shut down the ActorSystem. 
  system.shutdown()

}

2. STUDENTACTOR響應給INITSIGNAL消息并且發送一條QUOTEREQUEST消息給TEACHERACTOR

以及

4. STUDENTACTOR從TEACHERACTOR接收到QUOTERESPONSE并且將日志記錄到控制臺/日志記錄器

為什么我把2和4條綁在一起？因為實在太簡單了，要是拆開來你要恨死我。

所以，第二點 - StudentActor從DriverApp接到InitSignal消息并且發送QuoteRequest給TeacherActor。

def receive = {  
    case InitSignal=> {
          teacherActorRef!QuoteRequest
    }
    ...
    ...

就這么多！

第四點 - StudentActor記錄從TeacherActor發來的日志消息。

就像我們承諾的：

case QuoteResponse(quoteString) => {  
      log.info ("Received QuoteResponse from Teacher")
      log.info(s"Printing from Student Actor $quoteString")
}

我想你肯定同意現在這個看起來跟偽代碼一樣。

所以整個的StudentActor類看起來就是這樣：

STUDENTACTOR.SCALA

package me.rerun.akkanotes.messaging.requestresponse

import akka.actor.Actor  
import akka.actor.ActorLogging  
import me.rerun.akkanotes.messaging.protocols.TeacherProtocol._  
import me.rerun.akkanotes.messaging.protocols.StudentProtocol._  
import akka.actor.Props  
import akka.actor.ActorRef

class StudentActor (teacherActorRef:ActorRef) extends Actor with ActorLogging {

  def receive = {
    case InitSignal=> {
      teacherActorRef!QuoteRequest
    }

    case QuoteResponse(quoteString) => {
      log.info ("Received QuoteResponse from Teacher")
      log.info(s"Printing from Student Actor $quoteString")
    }
  }
}

3. TEACHERACTOR響應QUOTERESPONSE消息

這就跟我們在fire-n-forget 那里寫的代碼差不多。

TeacherActor接受一條QuoteRequest消息并且返回一條QuoteResponse。

TEACHERACTOR.SCALA

package me.rerun.akkanotes.messaging.requestresponse

import scala.util.Random

import akka.actor.Actor  
import akka.actor.ActorLogging  
import akka.actor.actorRef2Scala  
import me.rerun.akkanotes.messaging.protocols.TeacherProtocol._


class TeacherActor extends Actor with ActorLogging {

  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")

  def receive = {

    case QuoteRequest => {

      import util.Random

      //Get a random Quote from the list and construct a response
      val quoteResponse = QuoteResponse(quotes(Random.nextInt(quotes.size)))

      //respond back to the Student who is the original sender of QuoteRequest
      sender ! quoteResponse

    }
  }
}

測試用例TESTCASES

現在，我們的測試用例會模擬DriverApp。由于StudentActor只是記錄消息，我們無法對QuoteResponse消息進行斷言，我們只斷言消息在EventStream里出現（就像我們之前
說的）

所以，我們的測試用例就像這樣：

"A student" must {

    "log a QuoteResponse eventually when an InitSignal is sent to it" in {

      import me.rerun.akkanotes.messaging.protocols.StudentProtocol._

      val teacherRef = system.actorOf(Props[TeacherActor], "teacherActor")
      val studentRef = system.actorOf(Props(new StudentActor(teacherRef)), "studentActor")

      EventFilter.info (start="Printing from Student Actor", occurrences=1).intercept{
        studentRef!InitSignal
      }
    }
  }

源碼

整個工程可以在github下載。

下一步，我們會看到怎樣使用Akka的schedulers并且用Kamon監控你的Akka應用。

文章來自微信平臺「麥芽面包」,微信號「darkjune_think」。轉載請注明。