[翻译]AKKA笔记 - DEATHWATCH -7

当我们说Actor生命周期的时候,我们能看到Actor能被很多种方式停掉（用ActorSystem.stop或ActorContext.stop或发送一个PoisonPill - 也有一个kill和gracefulstop）。

无论Actor是怎么死的，有些情况一些系统中的其他actor想要知道。让我们举一个Actor与数据库交互的例子 - 我们叫它RepositoryActor。很明显，会有一些其他actor会向这个RepositoryActor发送消息。这些有“兴趣”的Actor很愿意留个eye或者看（watch）这个actor关闭时的消息。这个在Actor里面叫DeathWatch。这个用来watch和unwatch的方法就是ActorContext.watch和ActorContext.unwatch。如果已经监视了，这些watcher会在Actor关闭时收到一个Terminated消息，并可以很舒服的加到他们的receive功能中。

不像Supervision有一个严格的父子继承关系，任何Actor都可以watch任何ActorSystem中的Actor。

让我们看下。

代码

QUOTEREPOSITORYACTOR

1.我们的QueryRepositoryActor格言查询Actor保存了一个quote的列表并且在收到一个QuoteRepositoryRequest时随机返回一条。

1. 他记录了收到消息的个数，如果收到超过3个消息，他用PoisonPill把自己杀掉

这没啥神奇的。

package me.rerun.akkanotes.deathwatchimport akka.actor.{PoisonPill, Actor, ActorLogging, actorRef2Scala}  import me.rerun.akkanotes.protocols.QuoteRepositoryProtocol._  import scala.util.Randomclass QuoteRepositoryActor() 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")  var repoRequestCount:Int=1  def receive = {    case QuoteRepositoryRequest => {      if (repoRequestCount>3){        self!PoisonPill      }      else {        //Get a random Quote from the list and construct a response        val quoteResponse = QuoteRepositoryResponse(quotes(Random.nextInt(quotes.size)))        log.info(s"QuoteRequest received in QuoteRepositoryActor. Sending response to Teacher Actor $quoteResponse")        repoRequestCount=repoRequestCount+1        sender ! quoteResponse      }    }  }}

TEACHERACTORWATCHER

一样的，TeacherActorWatcher也没啥神奇的，除了他创建了一个QuoteRepositoryActor并且用context.watch观察。

package me.rerun.akkanotes.deathwatchimport akka.actor.{Terminated, Props, Actor, ActorLogging}  import me.rerun.akkanotes.protocols.TeacherProtocol.QuoteRequest  import me.rerun.akkanotes.protocols.QuoteRepositoryProtocol.QuoteRepositoryRequestclass TeacherActorWatcher extends Actor with ActorLogging {  val quoteRepositoryActor=context.actorOf(Props[QuoteRepositoryActor], "quoteRepositoryActor")  context.watch(quoteRepositoryActor)  def receive = {    case QuoteRequest => {      quoteRepositoryActor ! QuoteRepositoryRequest    }    case Terminated(terminatedActorRef)=>{      log.error(s"Child Actor {$terminatedActorRef} Terminated")    }  }}

测试CASE

这里会有点意思。我从来没想过这个可以被测试。akka-testkit。我们会分析下这三个测试CASE：

1. 断言如果观察到已经收到Terminated消息

QuoteRepositoryActor应该在收到第四条消息时给测试case发送一条Terminated消息。前三条应该是可以的。

"A QuoteRepositoryActor" must {    ...    ...    ...    "send back a termination message to the watcher on 4th message" in {      val quoteRepository=TestActorRef[QuoteRepositoryActor]      val testProbe=TestProbe()      testProbe.watch(quoteRepository) //Let's watch the Actor      within (1000 millis) {        var receivedQuotes = List[String]()        (1 to 3).foreach(_ => quoteRepository ! QuoteRepositoryRequest)        receiveWhile() {          case QuoteRepositoryResponse(quoteString) => {            receivedQuotes = receivedQuotes :+ quoteString          }        }        receivedQuotes.size must be (3)        println(s"receiveCount ${receivedQuotes.size}")        //4th message        quoteRepository!QuoteRepositoryRequest        testProbe.expectTerminated(quoteRepository)  //Expect a Terminated Message      }    }

2.如果没有观察（watched/unwatched）到则断言没收到Terminated消息

事实上，我们做这个只是演示下context.unwatch。如果我们移掉testProbe.watch和testProbe.unwatch这行，则测试case会运行的很正常。

    "not send back a termination message on 4th message if not watched" in {      val quoteRepository=TestActorRef[QuoteRepositoryActor]      val testProbe=TestProbe()      testProbe.watch(quoteRepository) //watching      within (1000 millis) {        var receivedQuotes = List[String]()        (1 to 3).foreach(_ => quoteRepository ! QuoteRepositoryRequest)        receiveWhile() {          case QuoteRepositoryResponse(quoteString) => {            receivedQuotes = receivedQuotes :+ quoteString          }        }        testProbe.unwatch(quoteRepository) //not watching anymore        receivedQuotes.size must be (3)        println(s"receiveCount ${receivedQuotes.size}")        //4th message        quoteRepository!QuoteRepositoryRequest        testProbe.expectNoMsg() //Not Watching. No Terminated Message      }    }

3. 在TeacherActorWatcher中断言收到了Terminated消息

我们订阅了EventStream并通过检查一个特殊的日志消息来断言termination。

   "end back a termination message to the watcher on 4th message to the TeacherActor" in {      //This just subscribes to the EventFilter for messages. We have asserted all that we need against the QuoteRepositoryActor in the previous testcase      val teacherActor=TestActorRef[TeacherActorWatcher]      within (1000 millis) {        (1 to 3).foreach (_=>teacherActor!QuoteRequest) //this sends a message to the QuoteRepositoryActor        EventFilter.error (pattern="""Child Actor .* Terminated""", occurrences = 1).intercept{          teacherActor!QuoteRequest //Send the dangerous 4th message        }      }    }

EventFilter中的pattern属性，没啥奇怪的，需要一个正则表达式。正则*pattern="""Child Actor . Terminated"""用来匹配一条格式是Child Actor {Actor[akka://TestUniversityMessageSystem/user/$$d/quoteRepositoryActor# -1905987636]} Terminated**日志信息。

Github

与往常一样，代码在github。看下deathwatch的包。

文章来自微信平台「麦芽面包」（微信扫描二维码关注）。未经允许，禁止转载。

关键字：akka

版权声明

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处。如若内容有涉嫌抄袭侵权/违法违规/事实不符，请点击 举报 进行投诉反馈！