[翻译]AKKA笔记 - DEATHWATCH -7

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

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

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

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

这里写图片描述

让我们看下。

代码

QUOTEREPOSITORYACTOR

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

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

这没啥神奇的。

package me.rerun.akkanotes.deathwatch

import akka.actor.{PoisonPill, Actor, ActorLogging, actorRef2Scala}  
import me.rerun.akkanotes.protocols.QuoteRepositoryProtocol._  
import scala.util.Random

class 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.deathwatch

import akka.actor.{Terminated, Props, Actor, ActorLogging}  
import me.rerun.akkanotes.protocols.TeacherProtocol.QuoteRequest  
import me.rerun.akkanotes.protocols.QuoteRepositoryProtocol.QuoteRepositoryRequest

class 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的包。


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

目录
相关文章
|
JSON 数据格式
Charles自动保存响应数据
Charles自动保存响应数据
1005 0
Charles自动保存响应数据
鲁棒优化入门(二)——基于matlab+yalmip求解鲁棒优化问题
鲁棒优化的含义就是在最恶劣的情况下(不确定变量的取值使目标函数最大),求出满足约束条件,并且能使目标函数最优的决策变量。 yalmip工具箱可以用来求解鲁棒优化问题,但还是有一定局限性的,并不能处理任意形式的不确定集下的鲁棒优化问题,一般来说,当鲁棒优化问题的不确定集合为箱型不确定集、椭球不确定集以及多面体不确定集时,都可以用yalmip工具箱求解(具体细节可参考官方文档)。本文介绍了利用yalmip求鲁棒优化问题的一般方法。......
|
5月前
|
人工智能 IDE 机器人
给阿里产品团队的真心话:千问很强,但我们需要它能“干活”!
作为一名一线程序员,建议千问尽快打通项目目录读取功能,并实现手机电脑端同步,做真正能干活的AI工具。
|
消息中间件 存储 NoSQL
RocketMQ实战—6.生产优化及运维方案
本文围绕RocketMQ集群的使用与优化,详细探讨了六个关键问题。首先,介绍了如何通过ACL配置实现RocketMQ集群的权限控制,防止不同团队间误用Topic。其次,讲解了消息轨迹功能的开启与追踪流程,帮助定位和排查问题。接着,分析了百万消息积压的处理方法,包括直接丢弃、扩容消费者或通过新Topic间接扩容等策略。此外,提出了针对RocketMQ集群崩溃的金融级高可用方案,确保消息不丢失。同时,讨论了为RocketMQ增加限流功能的重要性及实现方式,以提升系统稳定性。最后,分享了从Kafka迁移到RocketMQ的双写双读方案,确保数据一致性与平稳过渡。
|
8月前
|
前端开发 小程序 Docker
改论文熬到三点的“格式劫”:肝了几个通宵整出一个小工具,把豆包、DeepSeek、ChatGPT生成内容直接转成了可编辑Word
介绍了一款专为科研、教学人群打造的小工具的设计初衷和设计理念,轻松解决大模型生成内容转Word时的公式乱码、表格错乱、流程图模糊等“格式劫”。自动补全LaTeX符号、修复Markdown表格结构、高清渲染Mermaid图,一键输出可编辑、带样式的Word文档。无需安装、不用懂代码,复制粘贴即用,30秒完成繁琐排版,让写论文、出试卷不再熬夜改格式。
|
传感器
水传感器的技术原理有哪些
水传感器通过多种技术原理检测水质,包括电导率测量、光学感应、化学反应和生物传感等方法,可监测pH值、溶解氧、浊度等参数。
|
算法 计算机视觉 Python
YOLOv8优改系列二:YOLOv8融合ATSS标签分配策略,实现网络快速涨点
本文介绍了如何将ATSS标签分配策略融合到YOLOv8中,以提升目标检测网络的性能。通过修改损失文件、创建ATSS模块文件和调整训练代码,实现了网络的快速涨点。ATSS通过自动选择正负样本,避免了人工设定阈值,提高了模型效率。文章还提供了遇到问题的解决方案,如模块载入和环境配置问题。
1099 0
YOLOv8优改系列二:YOLOv8融合ATSS标签分配策略,实现网络快速涨点
|
消息中间件 监控 Java
您是否已集成 Spring Boot 与 ActiveMQ?
您是否已集成 Spring Boot 与 ActiveMQ?
643 0
|
Linux 网络安全 数据处理
【专栏】Linux下的xxd命令是一个强大的二进制数据处理工具,用于十六进制转储和数据分析,我教你应该如何使用!
【4月更文挑战第28天】Linux下的xxd命令是一个强大的二进制数据处理工具,用于十六进制转储和数据分析。它可以显示文件的十六进制和ASCII表示,方便查看内容、分析数据结构和比较文件。xxd支持指定输出格式、写入文件、数据提取和转换等功能。在网络安全分析、程序调试和数据恢复等领域有广泛应用。通过掌握xxd,用户能更深入理解和处理二进制数据。
2295 0
|
人工智能 决策智能
【AI Agent系列】【MetaGPT多智能体学习】3. 开发一个简单的多智能体系统,兼看MetaGPT多智能体运行机制
【AI Agent系列】【MetaGPT多智能体学习】3. 开发一个简单的多智能体系统,兼看MetaGPT多智能体运行机制
855 0