Akka 使用系列之三: 层次结构和容错机制

  我们需要实现一个翻译模块，其功能是输入中文输出多国语言。我们可以让一个 Master Actor 负责接收外界输入，多个 Worker Actor 负责将输入翻译成特定语言，Master Actor 和 Worker Actor 之间是上下级层次关系。下图展示了这种层级结构。

class Master extends Actor with ActorLogging{
    val english2chinese 
        = context.actorOf(Props[English2Chinese],"English2Chinese")
    val english2cat     
        = context.actorOf(Props[English2Cat],"English2Cat")

    def receive = {
        case eng1:String =>{
            english2chinese ! eng1
            english2cat     ! eng1
        }
    }
    
}

class English2Chinese extends Actor with ActorLogging{
    def receive = {
        case eng:String => {
            println("我翻译不出来!")
        }
    }
}

class English2Cat extends Actor with ActorLogging{
    def receive = {
        case eng:String =>{
             println( "喵喵喵!")
        }
    }
}

object Main{
    def main(args:Array[String]):Unit = {
        val sys = ActorSystem("system")
        val master = sys.actorOf(Props[Master],"Master")
        master ! "Hello,world!"
    }
}

  上面的 Actors 层次结构是我们程序里 Actor 的层次结构。这个层次结构是 Actor System 层次结构的一部分。Actor System 层次结构从根节点出来有两个子节点：UserGuardian 和 SystemGuardian。用户程序产生的所有 Actor 都在 UserGuardian 节点下，SystemGuardian 节点则包含系统中的一些 Actor，比如 deadLetterListener。如果一个 Actor 已经 stop 了，发送给这个 Actor 的消息就会被转送到 deadLetterListener。因此完整的 Actor 层次结构如下所示。

  对于分布式系统来说，容错机制是很重要的指标。那么 Akka 是怎么实现容错的呢？Akka 的容错机制是基于层次结构: Akka 在 Actor 加一个监控策略，对其子 Actor 进行监控。下面的代码是给 Actor 加了一个监控策略，其监控策略内容：如果子 Actor 在运行过程中抛出 Exception，对该子 Actor 执行停止动作 (即停止该子 Actor)。

 override val supervisorStrategy = OneForOneStrategy(){
      case _:Exception => Stop
}

  Akka 的监控策略一共支持四种动作：Stop, Resume, Restart 和 Escalate。

  Akka 的监控策略分为两种。一种是 OneForOne。这种策略只对抛出 Exception 的子 Actor 执行相应动作。还是拿上面的翻译模块做例子，我们加入一个 OneForOne 的 Stop 的监控策略。

class Master1 extends Actor with ActorLogging{
    val english2Chinese = 
    context.actorOf(Props[English2Chinese1],"English2Chinese")
    val english2Cat     = 
    context.actorOf(Props[English2Cat1], "English2Cat")
    override val supervisorStrategy = OneForOneStrategy(){
      case _:Exception                    => Stop
    }
    override def receive = {
      case eng:String => {
        english2Cat ! eng;
        english2Chinese ! eng;
      }
    }
}
class English2Chinese1 extends Actor with ActorLogging{
  override def receive = {
    case eng:String => {
      println("翻译不出来")
    }
  }
}
class English2Cat1 extends Actor with ActorLogging{
  override def receive = {
    case eng:String => {
      throw new Exception("Exception in English2Cat1")
    }
  }
}

object hierarchy1 {
  def main(args:Array[String]):Unit = {
    val system = ActorSystem("system")
    val master = system.actorOf(Props[Master1],"Master")
    master ! "Hello, world"
    Thread.sleep(1000)
    master ! "Hello, world"
  }
}

  另一种是 AllForOne。如果有子 Actor 抛出 Exception，这种监控策略对所有子 Actor 执行动作。

class Master2 extends Actor with ActorLogging{
    val english2Chinese = 
    context.actorOf(Props[English2Chinese2],"English2Chinese")
    val english2Cat     = 
    context.actorOf(Props[English2Cat2], "English2Cat")

    override val supervisorStrategy= AllForOneStrategy() {
      case _: Exception                   => Stop
    }

    override def receive = {
      case eng:String => {
        english2Cat ! eng;
        english2Chinese ! eng;
      }
    }
}

   我们使用 Akka 开发并行程序时，可以使用层级结构组织 Actors。层次结构不仅比较符合人类直觉，还为容错提供了机制保障。本文的所有代码已经上传到[ GitHub](https://github.com/algorithmdog/AkkaUsageLearner) 。欢迎关注 AlgorithmDog 公众号，每两周的更新会有推送哦。