Akka(6): become/unbecome:运算行为切换

   通过一段时间的学习了解,加深了一些对Akka的认识,特别是对于Akka在实际编程中的用途方面。我的想法,或者我希望利用Akka来达到的目的是这样的:作为传统方式编程的老兵,我们已经习惯了直线流程方式一口气实现完整的功能。如果使用Akka,我们可以把这个完整的功能分切成多个能产生中间临时结果的小功能然后把这些功能放到不同的Actor上分别独立运算,再通过消息来连接这些功能集合成最终结果。如此我们就轻易得到了一个多线程并发程序。由于Akka是软件工具(Tool),没有软件架构(Framework)对编程方式的特别要求,Actor的构建和使用非常方便,我们甚至不需要多少修改就可以直接把原来的一段代码移到Actor上。如果遇到一些重复的运算,我们还可以用Routing来实现并行运算。当然,把Actor当作简单的行令运算器可能还不够,如果能实现一些具体运算之上的高层次程序逻辑和流程就更加完善。我们可以用这样的高层次Actor去解析程序逻辑、执行流程、把具体的运算分配给其它各种运算Actor或者一组Routees并行运算从而取得整体程序的高效率运行。具备了这些功能后,也许我们就可以完全用Actor模式来替代传统单线程行令编程了。Akka可以通过Actor的动态行为转换来实现同一Actor在不同情况下提供不同的功能支持。我们前面提到Actor的功能是在receive函数内实现的。那么转换功能是否就是切换不同的receive函数呢?答案是确定的,Akka是通过Actor的context.become(rcvFunc)来实现receive函数切换的,我们看看下面这个示范:

import akka.actor._

object FillSeasons {
  case object HowYouFeel
  def props = Props(new FillSeasons)
}

class FillSeasons extends Actor with ActorLogging {
  import FillSeasons._

  override def receive: Receive = spring
  def Winter: Receive = {
    case HowYouFeel =>
      log.info("It's freezing cold!")
  }
  def summer: Receive = {
    case HowYouFeel =>
      log.info("It's hot hot hot!")
  }
  def spring: Receive = {
    case HowYouFeel =>
      log.info("It feels so goooood!")
  }
}

object Becoming extends App {
  val demoSystem = ActorSystem("demoSystem")

  val feelingsActor = demoSystem.actorOf(FillSeasons.props,"feelings")

  feelingsActor ! FillSeasons.HowYouFeel

}

在FeelingsActor里我们定义了三个receive函数,对共同的HowYouFeel消息采取了不同的反应。默认行为是spring。那么应该如何在三种行为中切换呢?用context.become(???),如下:

import akka.actor._

object FillSeasons {
  case object HowYouFeel
  case object ToSummer
  case object ToSpring
  case object ToWinter
  def props = Props(new FillSeasons)
}

class FillSeasons extends Actor with ActorLogging {
  import FillSeasons._

  override def receive: Receive = spring
  def winter: Receive = {
    case HowYouFeel =>
      log.info("It's freezing cold!")
    case ToSummer => context.become(summer)
    case ToSpring => context.become(spring)

  }
  def summer: Receive = {
    case HowYouFeel =>
      log.info("It's hot hot hot!")
    case ToSpring => context.become(spring)
    case ToWinter => context.become(winter)
  }
  def spring: Receive = {
    case HowYouFeel =>
      log.info("It feels so goooood!")
    case ToSummer => context.become(summer)
    case ToWinter => context.become(winter)
  }
}

object Becoming extends App {
  val demoSystem = ActorSystem("demoSystem")

  val feelingsActor = demoSystem.actorOf(FillSeasons.props,"feelings")

  feelingsActor ! FillSeasons.HowYouFeel
  feelingsActor ! FillSeasons.ToSummer
  feelingsActor ! FillSeasons.HowYouFeel
  feelingsActor ! FillSeasons.ToWinter
  feelingsActor ! FillSeasons.HowYouFeel
  feelingsActor ! FillSeasons.ToSpring
  feelingsActor ! FillSeasons.HowYouFeel

  scala.io.StdIn.readLine()
  demoSystem.terminate()
}

我们增加了三个消息来切换receive。运算结果如下:

[INFO] [06/08/2017 17:51:46.013] [demoSystem-akka.actor.default-dispatcher-3] [akka://demoSystem/user/feelings] It feels so goooood!
[INFO] [06/08/2017 17:51:46.019] [demoSystem-akka.actor.default-dispatcher-4] [akka://demoSystem/user/feelings] It's hot hot hot!
[INFO] [06/08/2017 17:51:46.028] [demoSystem-akka.actor.default-dispatcher-4] [akka://demoSystem/user/feelings] It's freezing cold!
[INFO] [06/08/2017 17:51:46.028] [demoSystem-akka.actor.default-dispatcher-4] [akka://demoSystem/user/feelings] It feels so goooood!


Process finished with exit code 0

就这样在几个receive里窜来窜去的好像已经能达到我们设想的目的了。看看Akka源代码中become和unbecome发现这样的做法是不正确的:

  def become(behavior: Actor.Receive, discardOld: Boolean = true): Unit =
    behaviorStack = behavior :: (if (discardOld && behaviorStack.nonEmpty) behaviorStack.tail else behaviorStack)

  def become(behavior: Procedure[Any]): Unit = become(behavior, discardOld = true)

  def become(behavior: Procedure[Any], discardOld: Boolean): Unit =
    become({ case msg ⇒ behavior.apply(msg) }: Actor.Receive, discardOld)

  def unbecome(): Unit = {
    val original = behaviorStack
    behaviorStack =
      if (original.isEmpty || original.tail.isEmpty) actor.receive :: emptyBehaviorStack
      else original.tail
  }

从上面的代码可以发现:调用become(x)实际上是把x压进了一个堆栈里。如果像我们这样不断调用become转来转去的,在堆栈上留下旧的行为函数实例最终会造成StackOverFlowError。所以Akka提供了unbecome,这是个堆栈弹出函数,把上一个become压进的行为函数再弹出来,释放一个堆栈空间。所以我们应该用unbecome来解决堆栈溢出问题。但是,如果在多个receive函数之间转换来实现行为变化的话,就难以正确掌握堆栈的压进,弹出冲抵配对,并且无法避免所谓的意大利面代码造成的混乱逻辑。所以,become/unbecome最好使用在两个功能之间的转换。我们再设计一个例子来示范:

sealed trait DBOperations
case class DBWrite(sql: String) extends DBOperations
case class DBRead(sql: String) extends DBOperations

sealed trait DBStates
case object Connected extends DBStates
case object Disconnected extends DBStates

DBoperations代表数据库读写操作。DBState代表数据库当前状态:连线Connected或断线Disconnected。只有数据库在Connected状态下才能进行数据库操作。顺理成章,我们需要两个receive函数:

import akka.actor._
sealed trait DBOperations
case class DBWrite(sql: String) extends DBOperations
case class DBRead(sql: String) extends DBOperations

sealed trait DBStates
case object Connected extends DBStates
case object Disconnected extends DBStates

object DBOActor {
  def props = Props(new DBOActor)
}

class DBOActor extends Actor with ActorLogging {

  override def receive: Receive = disconnected

  def disconnected: Receive = {
    case Connected =>
      log.info("Logon to DB.")
      context.become(connected)
  }
  def connected: Receive = {
    case Disconnected =>
      log.info("Logoff from DB.")
      context.unbecome()
    case DBWrite(sql) =>
      log.info(s"Writing to DB: $sql")
    case DBRead(sql) =>
      log.info(s"Reading from DB: $sql")
  }
}

object BecomeDB extends App {
  val dbSystem = ActorSystem("dbSystem")
  val dbActor = dbSystem.actorOf(DBOActor.props,"dbActor")

  dbActor ! Connected
  dbActor ! DBWrite("Update table x")
  dbActor ! DBRead("Select from table x")
  dbActor ! Disconnected

  scala.io.StdIn.readLine()
  dbSystem.terminate()

}

运算结果显示如下:

[INFO] [06/09/2017 11:44:40.093] [dbSystem-akka.actor.default-dispatcher-3] [akka://dbSystem/user/dbActor] Logon to DB.
[INFO] [06/09/2017 11:44:40.106] [dbSystem-akka.actor.default-dispatcher-3] [akka://dbSystem/user/dbActor] Writing to DB: Update table x
[INFO] [06/09/2017 11:44:40.107] [dbSystem-akka.actor.default-dispatcher-3] [akka://dbSystem/user/dbActor] Reading from DB: Select from table x
[INFO] [06/09/2017 11:44:40.107] [dbSystem-akka.actor.default-dispatcher-3] [akka://dbSystem/user/dbActor] Logoff from DB.

以上是按正确顺序向dbActor发出数据库操作指令后产生的结果。但是,我们是在一个多线程消息驱动的环境里。发送给dbActor的消息收到时间无法预料。我们试着调换一下指令到达顺序:

  dbActor ! DBWrite("Update table x")
  dbActor ! Connected
  dbActor ! DBRead("Select from table x")
  dbActor ! Disconnected

运算结果:

[INFO] [06/09/2017 11:54:57.264] [dbSystem-akka.actor.default-dispatcher-4] [akka://dbSystem/user/dbActor] Logon to DB.
[INFO] [06/09/2017 11:54:57.273] [dbSystem-akka.actor.default-dispatcher-4] [akka://dbSystem/user/dbActor] Reading from DB: Select from table x
[INFO] [06/09/2017 11:54:57.273] [dbSystem-akka.actor.default-dispatcher-4] [akka://dbSystem/user/dbActor] Logoff from DB.

漏掉了DBWrite操作。可以理解,所有connected状态之前的任何操作都不会真正生效。Akka提供了个Stash trait能把一个receive函数未处理的消息都存起来。然后用unstash()可以把存储的消息都转移到本Actor的邮箱里。我们可以用Stash来解决这个消息遗失问题:

  def disconnected: Receive = {
    case Connected =>
      log.info("Logon to DB.")
      context.become(connected)
      unstashAll()
    case _ => stash()
  }

所有消息遗失都是在Disconnected状态内发生的。在disconnected里我们用stash把所有非Connected消息存起来,然后在转换成Connected状态时把这些消息转到信箱。再看看运算结果:

object BecomeDB extends App {
  val dbSystem = ActorSystem("dbSystem")
  val dbActor = dbSystem.actorOf(DBOActor.props,"dbActor")


  dbActor ! DBWrite("Update table x")
  dbActor ! Connected
  dbActor ! DBRead("Select from table x")
  dbActor ! Disconnected

  scala.io.StdIn.readLine()
  dbSystem.terminate()

}

[INFO] [06/09/2017 12:01:54.518] [dbSystem-akka.actor.default-dispatcher-4] [akka://dbSystem/user/dbActor] Logon to DB.
[INFO] [06/09/2017 12:01:54.528] [dbSystem-akka.actor.default-dispatcher-4] [akka://dbSystem/user/dbActor] Writing to DB: Update table x
[INFO] [06/09/2017 12:01:54.528] [dbSystem-akka.actor.default-dispatcher-4] [akka://dbSystem/user/dbActor] Reading from DB: Select from table x
[INFO] [06/09/2017 12:01:54.528] [dbSystem-akka.actor.default-dispatcher-4] [akka://dbSystem/user/dbActor] Logoff from DB.

显示结果正确。下面就是整个示范的源代码:

import akka.actor._
sealed trait DBOperations
case class DBWrite(sql: String) extends DBOperations
case class DBRead(sql: String) extends DBOperations

sealed trait DBStates
case object Connected extends DBStates
case object Disconnected extends DBStates

object DBOActor {
  def props = Props(new DBOActor)
}

class DBOActor extends Actor with ActorLogging with Stash {

  override def receive: Receive = disconnected

  def disconnected: Receive = {
    case Connected =>
      log.info("Logon to DB.")
      context.become(connected)
      unstashAll()
    case _ => stash()
  }
  def connected: Receive = {
    case Disconnected =>
      log.info("Logoff from DB.")
      context.unbecome()
    case DBWrite(sql) =>
      log.info(s"Writing to DB: $sql")
    case DBRead(sql) =>
      log.info(s"Reading from DB: $sql")
  }
}

object BecomeDB extends App {
  val dbSystem = ActorSystem("dbSystem")
  val dbActor = dbSystem.actorOf(DBOActor.props,"dbActor")


  dbActor ! DBWrite("Update table x")
  dbActor ! Connected
  dbActor ! DBRead("Select from table x")
  dbActor ! Disconnected

  scala.io.StdIn.readLine()
  dbSystem.terminate()

}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏安恒网络空间安全讲武堂

nox&CSAW部分pwn题解

暑假的时候遇到了一群一起学习安全的小伙伴,在他们的诱劝下,开始接触国外的CTF比赛,作为最菜的pwn选手就试着先打两场比赛试试水,结果发现国外比赛真有意思哎嘿。

1303
来自专栏java工会

java调用webservice接口的几种方法

4174
来自专栏JackeyGao的博客

Django小技巧22: 设计一个好的模型

本篇将分享一些技巧,用户改进 Model 的设计。其中有很多与命名约定有关, 这可以大大的提高代码的可读性。

962
来自专栏郭霖

Android Volley完全解析(三),定制自己的Request

经过前面两篇文章的学习,我们已经掌握了Volley各种Request的使用方法,包括StringRequest、JsonRequest、ImageRequest...

2106
来自专栏蓝天

使用可重入函数进行更安全的信号处理

在早期的编程中,不可重入性对程序员并不构成威胁;函数不会有并发访问,也没有中断。在很多较老的 C 语言实现中,函数被认为是在单线程进程的环境中运行。

712
来自专栏JackieZheng

照虎画猫写自己的Spring

从细节跳出来 看了部分Spring的代码,前面用了四篇内容写了一些读书笔记。 回想起来,论复杂度,Spring够喝上好几壶的。他就像一颗枝繁叶茂的大树,远处看...

1878
来自专栏Hongten

java开发_UUID(Universally Unique Identifier,全局唯一标识符)和GUID(Globally Unique Identifier,全球唯一标识符)

GUID: 即Globally Unique Identifier(全球唯一标识符) 也称作 UUID(Universally Unique IDentifie...

901
来自专栏MasiMaro 的技术博文

ATL模板库中的OLEDB与ADO

上次将OLEDB的所有内容基本上都说完了,从之前的示例上来看OLEDB中有许多变量的定义,什么结果集对象、session对象、命令对象,还有各种缓冲等等,总体上...

1152
来自专栏与神兽党一起成长

jFinal路由解析源码分析

jFinal的路由解析是在JFinalFilter中做的,这个Filter也需要在web.xml中配置。JFinalFilter实现了javax.servlet...

1242
来自专栏IMWeb前端团队

webpack2 的 tree-shaking 好用吗?

代码压缩的现状 下面是一个使用 react 的业务的代码依赖,但是实际上业务代码中并没有对依赖图中标识的模块,也就是说构建工具将不需要的代码打包到了最终的代码当...

3495

扫码关注云+社区