Akka 使用系列之三: 層次結(jié)構(gòu)

作者：李立 2017-06-01 22:59:45

我們使用 Akka 開發(fā)并行程序時(shí)，可以使用層級結(jié)構(gòu)組織 Actors。層次結(jié)構(gòu)不僅比較符合人類直覺，還為容錯(cuò)提供了機(jī)制保障。

Akka 是用層次結(jié)構(gòu)組織 Actors 的。

1. Akka 的層次結(jié)構(gòu)

我們需要實(shí)現(xiàn)一個(gè)翻譯模塊，其功能是輸入中文輸出多國語言。我們可以讓一個(gè) Master Actor 負(fù)責(zé)接收外界輸入，多個(gè) Worker Actor 負(fù)責(zé)將輸入翻譯成特定語言，Master Actor 和 Worker Actor 之間是上下級層次關(guān)系。下圖展示了這種層級結(jié)構(gòu)。

具體代碼實(shí)現(xiàn)如下所示。

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])={ 
        val sys = ActorSystem("system") 
        val master = sys.actorOf(Props[Master],"Master") 
        master ! "Hello,world!" 
    } 
}

我們在 Master Actor 中使用 context.actorOf 實(shí)例化 English2Chinese 和 English2Cat，便可以在它們之間形成層次關(guān)系。這點(diǎn)通過它們的 actor 地址得到證實(shí)。

上面的 Actors 層次結(jié)構(gòu)是我們程序里 Actor 的層次結(jié)構(gòu)。這個(gè)層次結(jié)構(gòu)是 Actor System 層次結(jié)構(gòu)的一部分。Actor System 層次結(jié)構(gòu)從根節(jié)點(diǎn)出來有兩個(gè)子節(jié)點(diǎn)：UserGuardian 和 SystemGuardian。用戶程序產(chǎn)生的所有 Actor 都在 UserGuardian 節(jié)點(diǎn)下，SystemGuardian 節(jié)點(diǎn)則包含系統(tǒng)中的一些 Actor，比如 deadLetterListener。如果一個(gè) Actor 已經(jīng) stop 了，發(fā)送給這個(gè) Actor 的消息就會(huì)被轉(zhuǎn)送到 deadLetterListener。因此完整的 Actor 層次結(jié)構(gòu)如下所示。

2. Akka 的容錯(cuò)機(jī)制

對于分布式系統(tǒng)來說，容錯(cuò)機(jī)制是很重要的指標(biāo)。那么 Akka 是怎么實(shí)現(xiàn)容錯(cuò)的呢?Akka 的容錯(cuò)機(jī)制是基于層次結(jié)構(gòu): Akka 在 Actor 加一個(gè)監(jiān)控策略，對其子 Actor 進(jìn)行監(jiān)控。下面的代碼是給 Actor 加了一個(gè)監(jiān)控策略，其監(jiān)控策略內(nèi)容：如果子 Actor 在運(yùn)行過程中拋出 Exception，對該子 Actor 執(zhí)行停止動(dòng)作 (即停止該子 Actor)。

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

Akka 的監(jiān)控策略一共支持四種動(dòng)作：Stop, Resume, Restart 和 Escalate。

Stop：子 Actor 停止。
Resume：子 Actor 忽略引發(fā)異常的消息，繼續(xù)處理后續(xù)消息。
Restart：子 Actor 停止，重新初始化一個(gè)子 Actor 處理后續(xù)消息
Escalate：錯(cuò)誤太嚴(yán)重，自己已經(jīng)無法處理，將錯(cuò)誤信息上報(bào)給父 Actor。

Akka 的監(jiān)控策略分為兩種。一種是 OneForOne。這種策略只對拋出 Exception 的子 Actor 執(zhí)行相應(yīng)動(dòng)作。還是拿上面的翻譯模塊做例子，我們加入一個(gè) OneForOne 的 Stop 的監(jiān)控策略。

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])={ 
    val system  
    = ActorSystem("system") 
    val master 
    = system.actorOf(Props[Master1],"Master") 
 
    master ! "Hello, world" 
    Thread.sleep(1000) 
    master ! "Hello, world" 
  } 
}

運(yùn)行這段代碼，我們得到下面結(jié)果。從下面的結(jié)果，我們可以看出:***輪 English2Cat1 拋出了 Exception, English2Chinese1 正常工作;第二輪，English2Cat1 已經(jīng)死了，English2Chinese1 也已經(jīng)死亡了。這個(gè)結(jié)果說明監(jiān)控策略已經(jīng)將 MasterActor 的所有子 Actor 停止了。

另一種是 AllForOne。如果有子 Actor 拋出 Exception，這種監(jiān)控策略對所有子 Actor 執(zhí)行動(dòng)作。

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; 
     } 
  } 
}