互聯網 - 大數據

文章出處
本次內容可歸到社科經管人文類的，不涉及技術，起源于一些經歷與思考。
這次要談下價值觀，先看下維基定義：

價值觀 (Value)是一種處理事情判斷對錯、做選擇時取舍的標準。有益的事物才有正價值。對有益或有害的事物評判的標準就是一個人的價值觀。

這樣講太虛，其實所有國人從小就經歷了，任何一個國家的愛國主義教育都屬于這個范疇，有記憶沒？在想不起來只好給答案了-思想道德/政治課。這就是從娃娃抓起的一件事，有意義嗎？確實有意義，價值觀取向直接能跟國人素質掛鉤，如果一個人沒有了判斷對錯的準繩，結果就是遍地開花的“扶老人”案，人情冷漠；最近上線的“扶老人”保險又把等式證明了一次。
國家需要價值觀教育，執政黨的需要黨綱，指導黨員行為準則；社會主義國家軍隊由于受黨指揮，那么分散在各級指揮中心的政委，指導員必不可少，其主要的作用就是思想政治工作。
到企業呢？社會主義的國企都會有黨支部，保證國企受黨的指揮，定時傳達從中央過來的精神。例如像中移動這樣的運營商，每周都有公司內部發出的關于目前“三嚴三實”的主題文章。而這并不是社會主義國家專利，資本主義國家由于企業都是私人的，私人企業想要將最高層的思想傳達到全體員工，并希望讓全體員工按照公司想要的行為方式行事，這個就叫“企業文化”。
GE作為全球標桿企業，有很多對其研究介紹的書籍，包括各種杰克韋爾奇的傳記類書籍，其中都提到了其企業文化的塑造，其價值觀包括：

第一，以極大的熱情全力以赴地推動客戶成功。
第二，視“六個西格瑪”質量為生命，確保客戶永遠是第一受益者，用質量去推動增長。
第三，決不容忍官僚作風。因為官僚非常容易滋生，一不小心就會增加自己的人員，使機構變得臃腫。
第四，以無邊界工作方式行事，永遠尋找并應用最好的想法而無需在意其來源。
第五，重視全球智力資本及其提供者，我們鼓勵員工向我們提供新的主意、新的創意，然后鼓勵大家互相的分享。
第六，視變革為可以帶來增長的機會。
第七，確立一個明確、簡單和以客戶為核心的目標，并不斷更新和完善它的實施。
第八，創建一個“挑戰極限”的境界。

一個企業如果有一個明確的愿景，使命，價值觀，能讓人在最困難的時候能堅持下去；員工在做任何事的時候有一個判斷準繩，缺少這個，一些決策會議上無法取舍，到底要兼顧誰的利益？這個決策有沒有與公司使命相符？一切迎刃而解。
看不懂企業管理的人，會認為對員工搞企業文化是洗腦，耍流氓。縱觀世界500強企業，高凝聚力的企業哪個沒有企業文化？企業文化能保證篩掉不符合公司期望的人，讓有相同信念的人聚合，這就是一個企業家期望的最好結果。如我喜歡的大家都懂的手機界某“情懷”公司，能吸引到這家公司工作的人，我相信都秉承了相同的價值觀。
如果不同意，在企業管理的領域外，這事還有其他同義詞 - ** “信仰” **。
總有人說中國人沒有信仰，認為有了信仰，就沒那么多人做斷子絕孫的事。
那，你接受“洗腦”嗎？

---

文章來自微信平臺「麥芽面包」,微信號「darkjune_think」。轉載請注明。

文章出處
古典作為音樂的分支，是非常有趣的，這是我前幾年才發現的。為什么這么晚才發現呢？因為在這條路上一開始走錯了。
最早接觸古典樂還是個小學生，在那個年代，大家熱衷于上各種少年宮學習“才藝”，音樂那是必不可少的，但由于那時我在研修素描，沒啥多余精力。聽說古典樂不錯，就去市里唯一一個新華書店買了幾盤貝多芬的“命運”，“田園”。結果可想而知，小孩子哪聽得了那么沉重的音樂。那可是天天向上的少年啊，上來就“棒棒棒棒”， 那命運的敲門聲，毫無美感，一遍都沒聽完就扔一邊了。
其實初聽者最好還是別上交響樂，先嘗試下輕松的弦樂，如小提琴。我推薦下輕松愉快的維瓦爾第的四季（Le quattro stagioni），這個專輯在很多 影視作品里都有作為背景樂，開篇一聽就知道了，積極向上。維瓦爾第屬于巴洛克音樂流派，這種類型的音樂都會有點古代歐洲的宮廷氣，奢華與紙醉金迷兼具。
​
如果想聽點舒緩輕松的，莫扎特的就不錯，有一個專輯收集的一套曲目，在閑適的午后沐浴在陽光下欣賞，絕對睡著了。并且很多家長用莫扎特的音樂來胎教，據說能促進寶寶智力發展。 至于推薦，先聽聽他的小夜曲（Eine Kleine Nachtmusik）入門，其他的自己慢慢選聽就好。
隨著欣賞能力的慢慢提高，可以逐漸慢慢嘗試一些大型交響樂，這是一個逐漸適應的過程。腦力工作者在工作時聽一些流行樂或搖滾樂對大腦干擾太強，這類純音樂正合適，舒緩流暢，不會打斷大腦的工作流。
封面圖是DECCA廠出品的四季專輯封面。

---

文章來自微信平臺「麥芽面包」,微信號「darkjune_think」。轉載請注明。

文章出處
上次我們看Actor消息機制，我們看到開火-忘記型消息發出（意思是我們只要發個消息給Actor但是不期望有響應）。
技術上來講， 我們發消息給Actors就是要它的副作用。 這就是這么設計的。除了不響應， 目標Actor當然也可以對消息做以下事情-
1.發送一個響應給發送者（在我們的case里，TeacherActor應該響應一首名言給StudentActor）
2.轉發一個響應給其他的潛在的受眾Actor，他們也可能響應/轉發/產生副作用。Routers（路由）和Supervisors（監管）就是這個例子。（我們馬上就能看到）

---

REQUEST & RESPONSE
這次，我們只關注第一點- 請求-響應的周期。
圖片說明了我們這次在嘗試歸檔做什么。為了簡介，我沒有在圖中加入ActorSystem, Dispathcer或Mailboxes。

DriverApp發送一個InitSignal(初始化信號)消息給StudentActor.

StudentActor反饋給InitSignal消息并且發送一條QuoteRequest(格言請求)消息給TeacherActor。

TeacherActor，像我們在第一次看到的，響應一條QuoteResponse。

StudentActor只記錄QuoteResponse（格言響應）到控制臺/日志記錄器。

我們可以寫個testcase來驗證。
讓我們看下這4點的細節：
1. DRIVERAPP發送一個INITSIGNAL消息給STUDENTACTOR
至此，你可能已經猜出DriverApp會做什么。 就4件事：
1） 初始化ActorSystem
//Initialize the ActorSystem
val system = ActorSystem("UniversityMessageSystem")
2) 創建TeacherActor
//create the teacher actor
val teacherRef = system.actorOf(Props[TeacherActor], "teacherActor")
3) 創建StudentActor
//create the Student Actor - pass the teacher actorref as a constructor parameter to StudentActor
val studentRef = system.actorOf(Props(new StudentActor(teacherRef)), "studentActor")
你可以看到我把ActorRef的實例TeacherActor給StudentActor的構造函數，這樣StudentActor可以使用ActorRef發消息給TeacherActor。還有另外的方式達到這個目的（如傳入Props），但是這個方法對我們后面看到的Supervisors和Routers會方便點。我們也會即將看到child actors但現在還不是一個好方式 - Student創建Teacher聽起來怪怪的吧？
最后，
4） DriverApp會發一個InitSignal給StudentActor, z何陽StudentActor就能開始發送QuoteRequest（格言請求）給TeacherActor。
//send a message to the Student Actor
studentRef ! InitSignal
這個DriverClass講差不多了。 Thread.sleep和ActorSystem.shutdown在我們最后關閉ActorSystem的時候會在發消息時會等待幾秒保證結束。
** DRIVERAPP.SCALA**
package me.rerun.akkanotes.messaging.requestresponse
import akka.actor.ActorSystem
import akka.actor.Props
import me.rerun.akkanotes.messaging.protocols.StudentProtocol._
import akka.actor.ActorRef
object DriverApp extends App {
//Initialize the ActorSystem
val system = ActorSystem("UniversityMessageSystem")
//construct the teacher actor
val teacherRef = system.actorOf(Props[TeacherActor], "teacherActor")
//construct the Student Actor - pass the teacher actorref as a constructor parameter to StudentActor
val studentRef = system.actorOf(Props(new StudentActor(teacherRef)), "studentActor")
//send a message to the Student Actor
studentRef ! InitSignal
//Let's wait for a couple of seconds before we shut down the system
Thread.sleep(2000)
//Shut down the ActorSystem.
system.shutdown()
}
2. STUDENTACTOR響應給INITSIGNAL消息并且發送一條QUOTEREQUEST消息給TEACHERACTOR
以及
4. STUDENTACTOR從TEACHERACTOR接收到QUOTERESPONSE并且將日志記錄到控制臺/日志記錄器
為什么我把2和4條綁在一起？因為實在太簡單了，要是拆開來你要恨死我。
所以，第二點 - StudentActor從DriverApp接到InitSignal消息并且發送QuoteRequest給TeacherActor。
def receive = {
case InitSignal=> {
teacherActorRef!QuoteRequest
}
...
...
就這么多！
第四點 - StudentActor記錄從TeacherActor發來的日志消息。
就像我們承諾的：
case QuoteResponse(quoteString) => {
log.info ("Received QuoteResponse from Teacher")
log.info(s"Printing from Student Actor $quoteString")
}
我想你肯定同意現在這個看起來跟偽代碼一樣。
所以整個的StudentActor類看起來就是這樣：
STUDENTACTOR.SCALA
package me.rerun.akkanotes.messaging.requestresponse
import akka.actor.Actor
import akka.actor.ActorLogging
import me.rerun.akkanotes.messaging.protocols.TeacherProtocol._
import me.rerun.akkanotes.messaging.protocols.StudentProtocol._
import akka.actor.Props
import akka.actor.ActorRef
class StudentActor (teacherActorRef:ActorRef) extends Actor with ActorLogging {
def receive = {
case InitSignal=> {
teacherActorRef!QuoteRequest
}
case QuoteResponse(quoteString) => {
log.info ("Received QuoteResponse from Teacher")
log.info(s"Printing from Student Actor $quoteString")
}
}
}
3. TEACHERACTOR響應QUOTERESPONSE消息
這就跟我們在fire-n-forget 那里寫的代碼差不多。
TeacherActor接受一條QuoteRequest消息并且返回一條QuoteResponse。
TEACHERACTOR.SCALA
package me.rerun.akkanotes.messaging.requestresponse
import scala.util.Random
import akka.actor.Actor
import akka.actor.ActorLogging
import akka.actor.actorRef2Scala
import me.rerun.akkanotes.messaging.protocols.TeacherProtocol._
class TeacherActor 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")
def receive = {
case QuoteRequest => {
import util.Random
//Get a random Quote from the list and construct a response
val quoteResponse = QuoteResponse(quotes(Random.nextInt(quotes.size)))
//respond back to the Student who is the original sender of QuoteRequest
sender ! quoteResponse
}
}
}
測試用例TESTCASES
現在，我們的測試用例會模擬DriverApp。由于StudentActor只是記錄消息，我們無法對QuoteResponse消息進行斷言，我們只斷言消息在EventStream里出現（就像我們之前
說的）
所以，我們的測試用例就像這樣：
"A student" must {
"log a QuoteResponse eventually when an InitSignal is sent to it" in {
import me.rerun.akkanotes.messaging.protocols.StudentProtocol._
val teacherRef = system.actorOf(Props[TeacherActor], "teacherActor")
val studentRef = system.actorOf(Props(new StudentActor(teacherRef)), "studentActor")
EventFilter.info (start="Printing from Student Actor", occurrences=1).intercept{
studentRef!InitSignal
}
}
}
源碼
整個工程可以在github下載。
下一步，我們會看到怎樣使用Akka的schedulers并且用Kamon監控你的Akka應用。

---

文章來自微信平臺「麥芽面包」,微信號「darkjune_think」。轉載請注明。

文章出處
《摩托車修理店未來工作哲學》這本書是看到老羅推薦的，買了有段時間了，一直沒騰出來時間看，前兩天總算看完了。
作者的經歷很有意思，原本是個博士，畢業后在華盛頓一家智庫工作，干了半年后辭職開了一家摩托車修理店。這本書有一部分介紹了他對于目前高學歷教育下年輕人大學畢業后涌入供過于求的格子間白領工作市場現象的看法。
書的第二章開始就介紹了關于體力勞動和腦力勞動的問題。其中提到

腦力勞動與體力勞動的劃分并不是自然而然地形成的。
現代工業化設計將"所有可能的腦力工作都從車間轉移并集中到計劃或工藝部門的"。

中國在世界一體化經濟中不能幸免，政府花大力氣提高全民教育水平，所有的家庭都把孩子往大學送，這么多的人被強制要求在學校里坐在板凳上呆16年以上，當出來的時候，突然發現市場上并不需要這么多人來做事。書中提到了16年，我還真沒仔細想過以前在校園里呆了這么久，人類近代百年的精華被濃縮在十幾年里去學習，西方教育體系的設計人真是做了件偉大的工程。
回到白領這件事，白領的工作是相當抽象的，你會發現很難向上一輩解釋你的工作是什么，即使是同行業的不同工種，可能都不互相了解。 比如我以前就完全不明白網站運營是什么玩意，銷售和市場的區別。
軟件圈流傳的笑話，人家聽說你是學計算機的，就要求你幫忙推薦什么臺式機合算啊，買什么好啊，或者電腦壞了來修修啊。 就我當年在學校里的情況，至少有六七成以上的同學不懂這些，除非本身就喜歡數碼產品，有動手能力喜歡搗鼓電腦硬件的，這里還要強調一下是愛好搞電腦硬件的人。

真正的知識工作集中在少數精英手中。我們必須冷眼審視“知識工作”，不要相信未來就是腦力勞動的天下-那更可能是辦事員的天下。

很多手工勞動已變為程式化的，如工程領域工人并不需要知道how to build,只需要按照給的預制件拼裝即可。
作者講述了在維修工行業內，腦力勞動與體力勞動是分不開的，在定位問題時會遇到很多挑戰，每輛車的情況都不一樣，你并不知道車主或前車主對車做了什么，就算他們介紹了，大多也是半真半假的。這個情況很有趣，因為在維修的過程中很多事情取決于修理工的良心，一個有好奇心的修理工為了定位問題可能會在處理過程中損壞車主的部件，這個該由誰買單？
解決問題的樂趣同樣可以發生在診斷醫生與程序員身上，美劇house醫生就是醫生類問題的代表，這片子非常值得一看。程序員在修復軟件問題時也會遇到同樣的挑戰。在一個領域的專家與菜鳥的區別就是：

有些東西就擺在我的前面，但我缺乏發覺它們的相關知識。

一個人能否在自己的專業領域上走的更遠，書中也引用了另一本暢銷書《禪與摩托車維修藝術》的例子，那位作者在一次修摩托的過程中被修理店坑了三次，都是因為店里修理工的素質。以下引用下書中的小段內容：

雖然他們看起來很隨和，友善，輕松自在，卻沒有全心投入工作之中...他們對自己的工作沒有認同感，不會說“我是修理工”這種話。
...
他必須把摩托車的順暢運轉內化為熱心關切的目標。

這種情況太常見了，很多強調工作與生活平衡的人并沒有真的熱愛自己干的事，這樣是很遺憾的。國內的應試型教育體系造就了大批這樣的學生，從小到大沒有一件事情是自己做主的，報考學校，選擇專業，參加工作，各種人生決策都是家長做的。
由于并不真的對工作感興趣，每天上班混日子，大清早就想著下班，工資錢都花在下班后吃喝玩樂的年輕人相當多。認同這種生活狀態的人會信奉一句話“工作只是手段，而不是目的”，那么我們的目的又是什么呢？每個人確實都該思考下這個問題。
發散的太遠了，得收一收。這都是在看書后的一點看法，當然人生是多元的，誰的人生誰做主，但別被別人做了主。
希望每個人都能在有生之年找到自己真正喜歡的事情，并為之奮斗。

---

文章來自微信平臺「麥芽面包」
微信公眾號「darkjune_think」
轉載請注明。

文章出處

原文：http://rerun.me/2014/10/21/akka-notes-child-actors-and-path/

Actor是完全的繼承結構。你創建的任何Actor肯定都是一個其他Actor的child。
讓我們分析下：
PATH
我們用ActorSystem.actorof創建一個ActorRef并打印出他的path
val actorSystem=ActorSystem("SupervisionActorSystem")
val actorRef=actorSystem.actorOf(Props[BasicLifecycleLoggingTeacherActor])
println (actorRef.path) // (prints) akka://SupervisionActorSystem/user/$a
可以看到，一個path看起來很像是文件系統中的一個文件路徑。

這里的akka是固定的，因為這些都是Akka Actor的地址 - 與**file://**和**http://**前綴差不多（跟協議沒啥關系）

SupervisionActorSystem就是你創建的ActorSystem的名字。

user我們下節再說。

$a是系統為你生成出來的Actor的名字。你對操作系統給你隨機生成的文件名怎么看？很明顯是人都不喜歡，因為你之后還要用這個名字。所以，讓我們給他一個有意義的名字：

val actorRef=actorSystem.actorOf(Props[BasicLifecycleLoggingTeacherActor], "teacherActor")
println (actorRef.path) // (prints) akka://SupervisionActorSystem/user/teacherActor
現在這個路徑（path）看起來差不多了。
CHILD ACTORS
跟從ActorSystem里面創建的頂級actor類似，我們也可以給ActorContext創建child actor。事實上， Actor的容錯能力很大程度上就是靠使用Actor的繼承層次和一個parent管理child actor的生命周期的方式。
現在假設你又一個TeacherSupervisor并且你打算創建一個TeacherActor來作為Supervisor的child，可以用ActorContext.actorof來代替使用ActorSystem.actorof:
class TeacherSupervisor extends Actor with ActorLogging {
val teacherActor=context.actorOf(Props[TeacherActor], "teacherActor")
...
...
基本上，在任何應用里，不像頂層actor，你會創建一堆的child actor - 這意思就是與調用actorSystem.actorof不同，你會調用一堆actorContext.actor
你會注意到child actor的path是akka://SupervisionActorSystem/user/teacherSupervisor/teacherActor，看起來跟給父目錄創建一個子目錄是一樣的。
你什么時候開始創建子Actor？
在你的任務是由子任務或多個子任務組成的時候你就應該創建一個子actor了。在你執行的子任務是一個易錯的時候，你想要隔離他（這樣如果錯了，你能恢復他）的時候你也需要創建一個子actor。當task之間沒有父子關系時，你千萬別創建子actor。
并且，還可以讓子actor創建自己的子actor來代理他們自己的子任務。Actor的創建成本很低但產出卻很高（我們在談supervision的時候可以看到這個）
現在那個PATH中的USER是什么？
作為一個對比，我把ActorSystem比擬成一個Unix文件系統 - 有一個/root目錄，還有其他的/etc,/usr,/bin和其他目錄。
ActorSystem跟那個差不多。他創建一些頂層Actor - 最重要根Actor就是根目錄/， user Actor就是/usrr目錄，一個system Actor就是一個/system目錄。（還有一個/deadLetters來代表DeadLetterActorRef。我們在上一篇里面看到過）
ActorSystem內組合了三個Actor（從ActorRefProvider）。他們是ActorSystem創建的所有actor的根actor。

systemGuardian actor - 所有在/system下的actor的根

guardian actor - 所有/user下actor的根

rootGuardian Actor - systemGuardian和userGuardianactor
的根

/**
* Reference to the supervisor of guardian and systemGuardian; ....
*/
def rootGuardian: InternalActorRef
/**
* Reference to the supervisor used for all top-level user actors.
*/
def guardian: LocalActorRef
/**
* Reference to the supervisor used for all top-level system actors.
*/
def systemGuardian: LocalActorRef
/user(aka) user guardian
任何你在自己程序中像StudentActor或TeacherActor用ActorSystem的actof方法來創建的Actor都直接在/user。這就是之前teacherActor的路徑是/user/teacherActor的原因。
/system(aka) system guardian
當userGuardian死的時候system guardian會將自己關閉。當userGuardian關閉時這是合乎常理的， 他下面所有的業務actor都停掉了所以所有的管理員actor都需要一樣停掉。
我們能看到System Actor被創建在兩個地方 - 意思是在**/system*繼承關系下的actor。

像我們之前看到的，所有發給一個已經終結掉的Actor的消息都會被轉發給一個內部Actor（DeadLetterActor）的郵箱。DeadLetter Actor把每個消息包裝成**DeadLetter*然后發送給EventStream。另一個叫DeadLetterListener的Actor消費所有的DeadLetter并且將其作為一個日志消息發送出去。現在，DeadLetterListener是一個在/system/deadLetterListener下的system Actor。

想想我們之前寫的訂閱了EventStream的日志消息的TestEventListener?他們也是System actor。事實上，所有的akka.logger都是作為System actor來創建的。

class TeacherTest extends TestKit(ActorSystem("UniversityMessageSystem", ConfigFactory.parseString("""akka.loggers = ["akka.testkit.TestEventListener"]""")))
...
...
這個文檔提到所有用配置文件配置的Actor都會在啟動的時候被創建并部署到ActorSystem，躲在/system的保護傘下。當我找到有趣的地方再更新下這個。
/(aka)root guardian
像我們之前看到的，/下的Actor是user和system guardian的父 actor。

---

雜七雜八
技術上來講，root actor也有個父actor。這個actor的唯一任務就是當root actor失敗是關閉整個ActorSystem。因此他沒有被算在Actor的繼承結構里， Akka項目組叫他：
private[akka] val theOneWhoWalksTheBubblesOfSpaceTime: InternalActorRef = new MinimalActorRef {
...


---

文章來自微信平臺「麥芽面包」（微信掃描二維碼關注）。未經允許，禁止轉載。

文章出處
當我們說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時隨機返回一條。

他記錄了收到消息的個數，如果收到超過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的包。

---

文章來自微信平臺「麥芽面包」（微信掃描二維碼關注）。未經允許，禁止轉載。

文章出處
失敗更像是分布式系統的一個特性。因此Akka用一個容忍失敗的模型，在你的業務邏輯與失敗處理邏輯(supervision邏輯)中間你能有一個清晰的邊界。只需要一點點工作，這很贊。這就是我們要討論的主題。
ACTOR SUPERVISION
想象一個方法調用了你棧頂的方法但卻出了一個異常。那么在棧下的方法能做什么呢？

抓住異常并按順序處理恢復

抓住異常，也許記個日志并保持安靜。

下層的方法也可以選擇無視這個異常（或者抓住并重扔出來）

想象下一直扔到main方法仍然沒有處理這個異常。這種情況下，程序肯定會輸出一個異常給console后退出。
你可以把同樣的情況套用在線程上。如果一個子線程拋了異常而再假設run或**call*方法沒有處理它，那么這個異常就會期望放在父線程或主線程中解決，無論哪種情況，如果主線程沒有處理他，系統就會退出。
讓我們再看看 - 如果被context.actorof創建出來的子Actor因為一個異常失敗了。父actor（指supervisor）可以處理子actor的任何失敗。如果父actor做了，他可以處理并恢復（Restart/Resume）。另外，把異常傳遞（Escalate）給父actor。 還有一種做法，可以直接stop掉子actor - 這就是那個子actor結局了。 為什么我說父actor（那個supervisor）？這是因為akka的監護方式為家長監護 - 這意味著只有創建了actor的人才能監護他們。
就這么多了！我們已經覆蓋到所有監護指令（Directoives）了。
策略
我忘了說一點： 你已經知道一個Akka Actor可以創建子actor并且子actor也可以隨意創建他們自己的子actor。
現在，想下以下兩個場景：
1.OneForOneStrategy
你的actor創建了很多子actor并且每一個子actor都連接了不同的數據源。假設你運行的是一個將英語翻譯成多種語言的應用。
假設，一個子actor失敗了然而你可以接受在最終結果里跳過這個結果，你想怎么做？關掉這個服務？當然不，你可能想要重啟/關閉這個有問題的子actor。是吧？現在這個策略在Akka的監護策略中叫OneForOneStrategy策略 - 如果一個actor掛了，只單獨處理這個actor。
基于你的業務異常，你可能需要對不同的異常有不同的反應（停止，重啟，升級，恢復）。要配置你自己的策略，你只需要override你Actor類中的supervisorStrategy方法。
聲明OneForOneStrategy的例子
import akka.actor.Actor
import akka.actor.ActorLogging
import akka.actor.OneForOneStrategy
import akka.actor.SupervisorStrategy.Stop
class TeacherActorOneForOne extends Actor with ActorLogging {
...
...
override val supervisorStrategy=OneForOneStrategy() {
case _: MinorRecoverableException => Restart
case _: Exception => Stop
}
...
...
2.AllForOneStrategy策略
假設你在做一個外部排序 (這是個又證明了我沒啥創造力的例子！），你的每個塊都被一個不同的actor處理。突然，一個Actor失敗了并拋了一個異常。這樣再往下處理就沒什么意思了因為最終結果肯定是錯的。所以，邏輯就是停止stop所有的actor。
我為什么說stop而不是重啟？因為在這個例子里重啟也沒用，每個actor的mailbox在重啟時并不會被清理。所以，如果我們重啟了，另外的chunk仍然會被處理。這不是我們想要的。重建actor并用新的mailbox在這里是個更合適的策略。
與OneForOneStrategy一樣，只需要用AllForOneStrategy的實現覆寫supervisorStrategy
下面是例子
import akka.actor.{Actor, ActorLogging}
import akka.actor.AllForOneStrategy
import akka.actor.SupervisorStrategy.Escalate
import akka.actor.SupervisorStrategy.Stop
class TeacherActorAllForOne extends Actor with ActorLogging {
...
override val supervisorStrategy = AllForOneStrategy() {
case _: MajorUnRecoverableException => Stop
case _: Exception => Escalate
}
...
...
指令 DIRECTIVES
AllForOneStrategy和OneForOneStrategy的構造方法都接受一個叫Decider的PartialFunction[Throwable,Directive]方法，他把Throwable與Directive指令做了一個映射：
case _: MajorUnRecoverableException => Stop
這就簡單的四個指令 - Stop，Resume，Escalate和Restart
Stop
在異常發生時子actor會停止，任何發給停止的actor的消息都會被轉到deadLetter隊列。
Resume
子actor會忽略拋出異常的消息并且繼續處理隊列中的其他消息。
Restart
子actor會停止并且一個新的actor會初始化。繼續處理mailbox中其他的消息。世界對這個是無感知的因為同樣的ActorRef指向了新的Actor。
Escalate
supervisor復制了失敗并讓他的supervisor處理這個異常。
缺省策略
如果我們的actor沒指定任何策略但是創建了子actor。他們會怎樣處理？Actor會有一個缺省的策略：
override val supervisorStrategy=OneForOneStrategy() {
case _: ActorInitializationException=> Stop
case _: ActorKilledException => Stop
case _: DeathPactException => Stop
case _: Exception => Restart
}
所以，缺省策略處理了四個case：
1. ACTORINITIALIZATIONEXCEPTION => STOP
當actor不能初始化，他會拋出一個ActorInitializationException。actor會被停止。讓我們在preStart調用中模擬下這個：
package me.rerun.akkanotes.supervision
import akka.actor.{ActorSystem, Props}
import me.rerun.akkanotes.protocols.TeacherProtocol.QuoteRequest
import akka.actor.Actor
import akka.actor.ActorLogging
object ActorInitializationExceptionApp extends App{
val actorSystem=ActorSystem("ActorInitializationException")
val actor=actorSystem.actorOf(Props[ActorInitializationExceptionActor], "initializationExceptionActor")
actor!"someMessageThatWillGoToDeadLetter"
}
class ActorInitializationExceptionActor extends Actor with ActorLogging{
override def preStart={
throw new Exception("Some random exception")
}
def receive={
case _=>
}
}
Ru
運行ActorInitializationExceptionApp會產生一個ActorInitializationException 異常然后所有的消息都會進deadLetterActor的消息隊列：
Log
[ERROR] [11/10/2014 16:08:46.569] [ActorInitializationException-akka.actor.default-dispatcher-2] [akka://ActorInitializationException/user/initializationExceptionActor] Some random exception
akka.actor.ActorInitializationException: exception during creation
at akka.actor.ActorInitializationException$.apply(Actor.scala:164)
...
...
Caused by: java.lang.Exception: Some random exception
at me.rerun.akkanotes.supervision.ActorInitializationExceptionActor.preStart(ActorInitializationExceptionApp.scala:17)
...
...
[INFO] [11/10/2014 16:08:46.581] [ActorInitializationException-akka.actor.default-dispatcher-4] [akka://ActorInitializationException/user/initializationExceptionActor] Message [java.lang.String] from Actor[akka://ActorInitializationException/deadLetters] to Actor[akka://ActorInitializationException/user/initializationExceptionActor#-1290470495] was not delivered. [1] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

2. ACTORKILLEDEXCEPTION => STOP
當Actor被kill消息關閉后，他會拋出一個ActorKilledException。如果拋這個異常，缺省策略會讓子actor停止。看起來停止一個被kill掉的actor沒什么意義。但想想這個：

ActorKilledException 會被傳遞給supervisor。 那么之前我們在DeathWatch里面提到的Actor里的生命周期watch或deathwatchers。 直到Actor被停掉前watcher不會知道任何事情。

給Actor發送kill只會讓那個特定的監管actor知道。用stop處理會暫停那個actor的mailbox，暫停了子actor的mailbox，停止了子actor，發送了Terminated給所有子actor的watcher，發送給所有類一個Terminated，然后actor的watcher都會迅速失敗最終讓Actor自己停止、

package me.rerun.akkanotes.supervision
import akka.actor.{ActorSystem, Props}
import me.rerun.akkanotes.protocols.TeacherProtocol.QuoteRequest
import akka.actor.Actor
import akka.actor.ActorLogging
import akka.actor.Kill
object ActorKilledExceptionApp extends App{
val actorSystem=ActorSystem("ActorKilledExceptionSystem")
val actor=actorSystem.actorOf(Props[ActorKilledExceptionActor])
actor!"something"
actor!Kill
actor!"something else that falls into dead letter queue"
}
class ActorKilledExceptionActor extends Actor with ActorLogging{
def receive={
case message:String=> log.info (message)
}
}
Log
日志說只要ActorKilledException 進來，supervisor就會停掉actor并且消息會進入deadLetter隊列
INFO m.r.a.s.ActorKilledExceptionActor - something
ERROR akka.actor.OneForOneStrategy - Kill
akka.actor.ActorKilledException: Kill
INFO akka.actor.RepointableActorRef - Message [java.lang.String] from Actor[akka://ActorKilledExceptionSystem/deadLetters] to Actor[akka://ActorKilledExceptionSystem/user/$a#-1569063462] was not delivered. [1] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

3. DEATHPACTEXCEPTION => STOP
在DeathWatch文中，你可以看到當一個Actor觀察一個子Actor時，他期望在他的receive中處理Terminated消息。如果沒有呢？你會得到一個DeathPactException
代碼演示了supervisorwatch子actor但沒有從子actor處理Terminated消息。
package me.rerun.akkanotes.supervision
import akka.actor.{ActorSystem, Props}
import me.rerun.akkanotes.protocols.TeacherProtocol.QuoteRequest
import akka.actor.Actor
import akka.actor.ActorLogging
import akka.actor.Kill
import akka.actor.PoisonPill
import akka.actor.Terminated
object DeathPactExceptionApp extends App{
val actorSystem=ActorSystem("DeathPactExceptionSystem")
val actor=actorSystem.actorOf(Props[DeathPactExceptionParentActor])
actor!"create_child" //Throws DeathPactException
Thread.sleep(2000) //Wait until Stopped
actor!"someMessage" //Message goes to DeadLetters
}
class DeathPactExceptionParentActor extends Actor with ActorLogging{
def receive={
case "create_child"=> {
log.info ("creating child")
val child=context.actorOf(Props[DeathPactExceptionChildActor])
context.watch(child) //Watches but doesnt handle terminated message. Throwing DeathPactException here.
child!"stop"
}
case "someMessage" => log.info ("some message")
//Doesnt handle terminated message
//case Terminated(_) =>
}
}
class DeathPactExceptionChildActor extends Actor with ActorLogging{
def receive={
case "stop"=> {
log.info ("Actor going to stop and announce that it's terminated")
self!PoisonPill
}
}
}
Log
日志告訴我們DeathPactException 進來了，supervisor停止了actor然后消息都進入了deadLetter的隊列
INFO m.r.a.s.DeathPactExceptionParentActor - creating child
INFO m.r.a.s.DeathPactExceptionChildActor - Actor going to stop and announce that it's terminated
ERROR akka.actor.OneForOneStrategy - Monitored actor [Actor[akka://DeathPactExceptionSystem/user/$a/$a#-695506341]] terminated
akka.actor.DeathPactException: Monitored actor [Actor[akka://DeathPactExceptionSystem/user/$a/$a#-695506341]] terminated
INFO akka.actor.RepointableActorRef - Message [java.lang.String] from Actor[akka://DeathPactExceptionSystem/deadLetters] to Actor[akka://DeathPactExceptionSystem/user/$a#-1452955980] was not delivered. [1] dead letters encountered. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

4. EXCEPTION => RESTART
對于其他的異常，缺省的指令是重啟Actor。看下這個應用。只是要證明下Actor重啟了，OtherExceptionParentActor讓child拋出一個異常并立刻發送一條消息。消息在子actor重啟的時候到達了mailbox，并被處理了。真不錯！
[圖片描述][5]
package me.rerun.akkanotes.supervision
import akka.actor.Actor
import akka.actor.ActorLogging
import akka.actor.ActorSystem
import akka.actor.OneForOneStrategy
import akka.actor.Props
import akka.actor.SupervisorStrategy.Stop
object OtherExceptionApp extends App{
val actorSystem=ActorSystem("OtherExceptionSystem")
val actor=actorSystem.actorOf(Props[OtherExceptionParentActor])
actor!"create_child"
}
class OtherExceptionParentActor extends Actor with ActorLogging{
def receive={
case "create_child"=> {
log.info ("creating child")
val child=context.actorOf(Props[OtherExceptionChildActor])
child!"throwSomeException"
child!"someMessage"
}
}
}
class OtherExceptionChildActor extends akka.actor.Actor with ActorLogging{
override def preStart={
log.info ("Starting Child Actor")
}
def receive={
case "throwSomeException"=> {
throw new Exception ("I'm getting thrown for no reason")
}
case "someMessage" => log.info ("Restarted and printing some Message")
}
override def postStop={
log.info ("Stopping Child Actor")
}
}
Log
1.異常拋出了，我們能在trace中看到

子類重啟了 - stop和start被調用了（我們稍后能看到preRestart和postRestart）

消息在重啟開始前被發送給子actor。

INFO m.r.a.s.OtherExceptionParentActor - creating child
INFO m.r.a.s.OtherExceptionChildActor - Starting Child Actor
ERROR akka.actor.OneForOneStrategy - I'm getting thrown for no reason
java.lang.Exception: I'm getting thrown for no reason
at me.rerun.akkanotes.supervision.OtherExceptionChildActor$$anonfun$receive$2.applyOrElse(OtherExceptionApp.scala:39) ~[classes/:na]
at akka.actor.Actor$class.aroundReceive(Actor.scala:465) ~[akka-actor_2.11-2.3.4.jar:na]
...
...
INFO m.r.a.s.OtherExceptionChildActor - Stopping Child Actor
INFO m.r.a.s.OtherExceptionChildActor - Starting Child Actor
INFO m.r.a.s.OtherExceptionChildActor - Restarted and printing some Message
ESCALATE AND RESUME
我們在defaultStrategy中看到stop和restart的例子。現在讓我們快速看下Escalate。
Resume忽略了異常并處理mailbox中的下條消息。這就像是抓住了異常但什么事也沒做。
Escalating更像是異常是致命的而supervisor不能處理它。所以，他要向他的supervisor求救。讓我們看個例子。
假設有三個Actor - EscalateExceptionTopLevelActor, EscalateExceptionParentActor 和 EscalateExceptionChildActor。 如果一個子actor拋出一個日常并且父級別actor不能處理它，他可以Escalate這個異常到頂級actor。頂級actor也可以選擇對哪些指令做出響應。在我們的例子里，我們只是做了stop。stop會立即停掉child（這里是EscalateExceptionParentActor）。我們知道，當一個actor執行stop時，他的所有子類都會在actor自己停掉前先停止。
package me.rerun.akkanotes.supervision
import akka.actor.Actor
import akka.actor.ActorLogging
import akka.actor.ActorSystem
import akka.actor.OneForOneStrategy
import akka.actor.Props
import akka.actor.SupervisorStrategy.Escalate
import akka.actor.SupervisorStrategy.Stop
import akka.actor.actorRef2Scala
object EscalateExceptionApp extends App {
val actorSystem = ActorSystem("EscalateExceptionSystem")
val actor = actorSystem.actorOf(Props[EscalateExceptionTopLevelActor], "topLevelActor")
actor ! "create_parent"
}
class EscalateExceptionTopLevelActor extends Actor with ActorLogging {
override val supervisorStrategy = OneForOneStrategy() {
case _: Exception => {
log.info("The exception from the Child is now handled by the Top level Actor. Stopping Parent Actor and its children.")
Stop //Stop will stop the Actor that threw this Exception and all its children
}
}
def receive = {
case "create_parent" => {
log.info("creating parent")
val parent = context.actorOf(Props[EscalateExceptionParentActor], "parentActor")
parent ! "create_child" //Sending message to next level
}
}
}
class EscalateExceptionParentActor extends Actor with ActorLogging {
override def preStart={
log.info ("Parent Actor started")
}
override val supervisorStrategy = OneForOneStrategy() {
case _: Exception => {
log.info("The exception is ducked by the Parent Actor. Escalating to TopLevel Actor")
Escalate
}
}
def receive = {
case "create_child" => {
log.info("creating child")
val child = context.actorOf(Props[EscalateExceptionChildActor], "childActor")
child ! "throwSomeException"
}
}
override def postStop = {
log.info("Stopping parent Actor")
}
}
class EscalateExceptionChildActor extends akka.actor.Actor with ActorLogging {
override def preStart={
log.info ("Child Actor started")
}
def receive = {
case "throwSomeException" => {
throw new Exception("I'm getting thrown for no reason.")
}
}
override def postStop = {
log.info("Stopping child Actor")
}
}
Log
可以在log中看到,

子actor拋了異常。

supervisor（EscalateExceptionParentActor）升級了（escalate）異常拋給了他的supervisor（EscalateExceptionTopLevelActor）

EscalateExceptionTopLevelActor 的指令是關閉actor。在順序上，子actor先停止。

父actor之后再停止（在watcher被通知后）

INFO m.r.a.s.EscalateExceptionTopLevelActor - creating parent
INFO m.r.a.s.EscalateExceptionParentActor - Parent Actor started
INFO m.r.a.s.EscalateExceptionParentActor - creating child
INFO m.r.a.s.EscalateExceptionChildActor - Child Actor started
INFO m.r.a.s.EscalateExceptionParentActor - The exception is ducked by the Parent Actor. Escalating to TopLevel Actor
INFO m.r.a.s.EscalateExceptionTopLevelActor - The exception from the Child is now handled by the Top level Actor. Stopping Parent Actor and its children.
ERROR akka.actor.OneForOneStrategy - I'm getting thrown for no reason.
java.lang.Exception: I'm getting thrown for no reason.
at me.rerun.akkanotes.supervision.EscalateExceptionChildActor$$anonfun$receive$3.applyOrElse(EscalateExceptionApp.scala:71) ~[classes/:na]
...
...
INFO m.r.a.s.EscalateExceptionChildActor - Stopping child Actor
INFO m.r.a.s.EscalateExceptionParentActor - Stopping parent Actor
請記住無論哪個指令發出只會使用在被escalated的子類上。 例如，一個restart指令從頂層發出，只有父類會被重啟并且在構造函數中/preStart中的都會被執行。如果一個父actor的子類在構造函數中唄創建，他們就會被創建。然而，在消息中創建的child的父actor仍然會在Terminated狀態。
TRIVIA
實際上，你可以控制是否preStart被調用。我們可以在下節看到。如果你好奇，可以看下Actor中的**postRestart*方法
def postRestart(reason: Throwable): Unit = {
preStart()
}
代碼
跟往常一樣，代碼在github

---

文章來自微信平臺「麥芽面包」（微信掃描二維碼關注）。未經允許，禁止轉載。

文章出處

原文地址：http://rerun.me/2016/05/21/akka-notes-finite-state-machines-1/

我最近有個機會在工作上使用了Akka FSM，是個非常有趣的例子。API（實際上就是DSL），使用體驗很棒。這里是我嘗試用Akka FSM的有限狀態機來寫日志。作為例子，我們會以構建一個咖啡機的步驟作為例子。
為什么不用BECOME和UNBECOME
我們知道plain vanilla Akka Actor可以用become/unbecome切換行為。那么，為什么我們需要Akka FSM？不能簡單點用Actor在狀態間切換？ 當然可以。但是當Akka的become and unbecome被一堆狀態攪在一起并不停地切換狀態的時候，建一個有許多狀態的狀態機能讓代碼迅速變的異常難懂（并且難調試）。
沒啥奇怪的，常見的建議就是當你在Actor時使用超過2種狀態就切換到Akka FSM。
AKKA FSM是啥
Akka FSM是Akka用來簡化管理Actor中不同狀態和切換狀態而構建有限狀態機的方法。
在底層，Akka FSM就是一個繼承了Actor的trait。
trait FSM[S, D] extends Actor with Listeners with ActorLogging
FSM trait提供的是純魔法 - 他提供了一個包裝了常規Actor的DSL，讓我們能集中注意力在更快的構建手頭的狀態機上。
換句話說，我們的常規Actor只有一個receive方法，FSM trait包裝了receive方法的實現并將調用指向到一個特定狀態機的處理代碼塊。
在我寫完代碼后注意的另一個事，就是完整的FSM Actor仍然很干凈并易懂。

---

現在讓我們開始看代碼。之前說過，我們要用Akka FSM建一個咖啡機。狀態機是這樣的：
狀態和數據
在FSM中，有兩個東西是一直存在的 - 任何時間點都有狀態 ，和在狀態中進行共享的數據。 在Akka FSM，想要校驗哪個是自己的數據，哪個是狀態機的數據，我們只要檢查這個聲明。
class CoffeeMachine extends FSM[MachineState, MachineData]
這代表所有的fsm的狀態繼承自MachineState，而所有在狀態間共享的數據就是MachineData。
作為一種風格，跟普通Actor一樣我們在companion對象中聲明所有的消息，所以我們在companion對象中聲明了狀態和數據：
object CoffeeMachine {
sealed trait MachineState
case object Open extends MachineState
case object ReadyToBuy extends MachineState
case object PoweredOff extends MachineState
case class MachineData(currentTxTotal: Int, costOfCoffee: Int, coffeesLeft: Int)
}
在狀態機的圖中，我們有三個狀態 - 打開，可買和關閉。 我們的數據，MachineData保留了開飛機關閉前機器中咖啡的數量（coffeesLeft），每杯咖啡的價格（costOfCoffee），咖啡機存放的零錢（currentTxTotal） - 如果零錢比咖啡價格低，機器就不賣咖啡，如果多，那么我們能找回零錢。
關于狀態和數據就這么多了。
在我們看每個狀態機的實現和用戶可用狀態機做的交互前， 我們先在5萬英尺看下FSM Actor。
FSM ACTOR的結構
FSM Actor的結構看起來跟我們的狀態機圖的差不多：
class CoffeeMachine extends FSM[MachineState, MachineData] {
//What State and Data must this FSM start with (duh!)
startWith(Open, MachineData(..))
//Handlers of State
when(Open) {
...
...
when(ReadyToBuy) {
...
...
when(PoweredOff) {
...
...
//fallback handler when an Event is unhandled by none of the States.
whenUnhandled {
...
...
//Do we need to do something when there is a State change?
onTransition {
case Open -> ReadyToBuy => ...
...
...
}
我們能從結構中看出什么：
1）我們有一個初始狀態（Open），when（open）代碼塊處理Open狀態的
收到的消息，ReadyToBuy狀態由when（ReadyToBuy）代碼塊來處理。我提到的消息與常規我們發給Actor的消息時一樣的，消息與數據一起包裝過。包裝后的叫做Event（akka.actor.FSM.Event），看起來的樣例是這樣Event(deposit: Deposit, MachineData(currentTxTotal, costOfCoffee, coffeesLeft))
Akka的文檔介紹：
/**
* All messages sent to the [[akka.actor.FSM]] will be wrapped inside an
* `Event`, which allows pattern matching to extract both state and data.
*/
case class Event[D](event: Any, stateData: D) extends NoSerializationVerificationNeeded
2）我們還能看到when方法接受兩個參數 - 第一個是狀態的名字，如Open,ReadyToBuy，另一個參數是PartialFunction， 與Actor的receive方法一樣做模式匹配。最重要的事是每一個模式匹配的case塊必須返回一個狀態（下次會講）。所以，代碼塊會是這樣的
when(Open) {
case Event(deposit: Deposit, MachineData(currentTxTotal, costOfCoffee, coffeesLeft)) => {
...
...
3)基本上， 消息中匹配到了when中第二個參數的模式會被一個特定狀態來處理。如果沒有匹配到，FSM Actor會嘗試將我們的消息與whenUnhandled塊中的模式進行匹配。理論上，所有在模式中沒有匹配到的消息都會被whenUnhandled處理。(我倒不太想建議編碼風格不過你可以聲明小點的PartialFunction并用andThen組合使用它，這樣你就能在選好的狀態中重用模式匹配。）
4）最后，還有個onTransition方法能讓你在狀態變化時做出反應或得到通知。
交互/消息
會有兩類人與咖啡機交互，喝咖啡的人，需要咖啡和咖啡機，和維護咖啡機做管理工作的人。
為了便于管理，所有與機器的交互里我用了兩個trait。（再提一下，一個交互/消息是與MachineData一起并被包在Event中的第一個元素。在原來的老Actor協議中，這個與發消息給Actor是一樣的。
object CoffeeProtocol {
trait UserInteraction
trait VendorInteraction
...
...
供應商交互
讓我們也聲明一下供應商可以與機器做的交互。
case object ShutDownMachine extends VendorInteraction
case object StartUpMachine extends VendorInteraction
case class SetCostOfCoffee(price: Int) extends VendorInteraction
//Sets Maximum number of coffees that the vending machine could dispense
case class SetNumberOfCoffee(quantity: Int) extends VendorInteraction
case object GetNumberOfCoffee extends VendorInteraction
所以，供應商可以

打開或關閉機器

設置咖啡的價格

設置和拿到機器中已有咖啡的數量。

用戶交互
case class Deposit(value: Int) extends UserInteraction
case class Balance(value: Int) extends UserInteraction
case object Cancel extends UserInteraction
case object BrewCoffee extends UserInteraction
case object GetCostOfCoffee extends UserInteraction
那么，對于用戶交互， 用戶可以

存錢買一杯咖啡

如果錢比咖啡的價格高那么可以得到找零

如果存的錢正好或高于咖啡價格機器就可以讓咖啡機做咖啡

在煮咖啡前取消交易過程并拿到所有的退款

問機器查詢咖啡的價格

下一篇，我們會看下所有的狀態并研究下他們的交互。
代碼
代碼在github.

---

文章來自微信平臺「麥芽面包」
微信公眾號「darkjune_think」轉載請注明。
如果覺得有趣，微信掃一掃關注公眾號。

文章出處
AKKA 筆記 - 有限狀態機 -2

原文地址： http://rerun.me/2016/05/22/akka-notes-finite-state-machines-2/

在上一節的Akka FSM筆記中，我們看了一些基本的使用Akka FSM和咖啡機的使用方式 - Actor的數據結構和一隊我們要發給Actor的消息。這次的第二部分也是最終部分，我們會過一遍這些狀態的實現細節。
總結
作為一個快速的總結，讓我們先看一下FMS的結構和我們要發過去的消息。
狀態和數據
FSM的三個狀態和要在各個狀態發送的數據是：
object CoffeeMachine {
sealed trait MachineState
case object Open extends MachineState
case object ReadyToBuy extends MachineState
case object PoweredOff extends MachineState
case class MachineData(currentTxTotal: Int, costOfCoffee: Int, coffeesLeft: Int)
}
消息
我們發給FSM的咖啡機和用戶交互的消息是：
object CoffeeProtocol {
trait UserInteraction
trait VendorInteraction
case class Deposit(value: Int) extends UserInteraction
case class Balance(value: Int) extends UserInteraction
case object Cancel extends UserInteraction
case object BrewCoffee extends UserInteraction
case object GetCostOfCoffee extends UserInteraction
case object ShutDownMachine extends VendorInteraction
case object StartUpMachine extends VendorInteraction
case class SetNumberOfCoffee(quantity: Int) extends VendorInteraction
case class SetCostOfCoffee(price: Int) extends VendorInteraction
case object GetNumberOfCoffee extends VendorInteraction
case class MachineError(errorMsg:String)
}
FSM ACTOR的結構
這是我們在第一節看到的大致結構：
class CoffeeMachine extends FSM[MachineState, MachineData] {
//What State and Data must this FSM start with (duh!)
startWith(Open, MachineData(..))
//Handlers of State
when(Open) {
...
...
when(ReadyToBuy) {
...
...
when(PoweredOff) {
...
...
//fallback handler when an Event is unhandled by none of the States.
whenUnhandled {
...
...
//Do we need to do something when there is a State change?
onTransition {
case Open -> ReadyToBuy => ...
...
...
}
狀態初始化
跟其他狀態機一樣， FSM在啟動時需要一個初始化狀態。這個可以在Akka FSM內聲明一個叫startWith的方法來實現。startWith接受兩個參數 - 初始化狀態和初始化數據。
class CoffeeMachine extends FSM[MachineState, MachineData] {
startWith(Open, MachineData(currentTxTotal = 0, costOfCoffee = 5, coffeesLeft = 10))
...
...
以上代碼說明了FSM的初始化狀態是Open并且當咖啡機Open時的初始化數據是
MachineData(currentTxTotal = 0, costOfCoffee = 5, coffeesLeft = 10).
當機器啟動時，咖啡機是一個干凈的狀態。它跟用戶還沒有任何交互，當前的余額是0。咖啡的價格唄設置成5元，總共能提供的咖啡設置為10杯。當咖啡機沖了10杯咖啡后數量為0時，咖啡機會shut down。
狀態的實現
終于到最后了！！
我覺得最簡單的方式來看咖啡機狀態的交互就是給交互做個分組，為FSM的實現寫測試用例。
如果你看下github的代碼，所有的測試用例都在CoffeeSpec并且FSM在CoffeeMachine
以下所有的測試都被CoffeeSpec測試類包裝了，聲明就像這樣：
class CoffeeSpec extends TestKit(ActorSystem("coffee-system")) with MustMatchers with FunSpecLike with ImplicitSender
設置并得到咖啡的價格
像我們之前看到的，MachineData初始化時設置為每杯咖啡5元并總數為10杯。這只是一個初始狀態，咖啡機必須能在任何時候設置咖啡的價格和能提供的數量。
通過發送SetCostOfCoffee消息給Actor可以設置價格。我們也應該能拿到咖啡的價格。這個可以通過發送GetCostOfCoffee消息給機器來獲得。
測試用例
describe("The Coffee Machine") {
it("should allow setting and getting of price of coffee") {
val coffeeMachine = TestActorRef(Props(new CoffeeMachine()))
coffeeMachine ! SetCostOfCoffee(7)
coffeeMachine ! GetCostOfCoffee
expectMsg(7)
}
...
...
...
實現
像我們在第一節討論的，所有發給FSM的消息都被包裝成Event類，并且也被MachineData包裝：
when(Open) {
case Event(SetCostOfCoffee(price), _) => stay using stateData.copy(costOfCoffee = price)
case Event(GetCostOfCoffee, _) => sender ! (stateData.costOfCoffee); stay()
...
...
}
}

以上代碼有幾個新詞 - stay,using和stateData,讓我們下面看下。
STAY和GOTO
想法是每一個被阻塞的case都必須返回一個State。這個可以用stay來完成，含義是已經在處理這條消息的最后了（SetCostOfCoffee或GetCostOfCoffee),咖啡機還在用一個狀態，在這里是Open狀態。
goto， 將狀態變為另一個。我們在討論Deposit時能看到它是怎么做的。
沒啥奇怪的，看下stay方法的實現：
final def stay(): State = goto(currentState.stateName)
USING
你可能已經猜到了，using方法可以讓我們把改過的數據傳給下個狀態。在SetCostOfCoffee消息的例子里，我們設置了MachineData的costOfCoffee域。由于狀態是個用例的例子（強烈建議使用不可變除非你喜歡debug），我們做了個copy。
狀態數據STATEDATA
stateData是一個我們用來操作FSM數據的方法，就是MachineData。 所以，以下代碼塊是等價的
case Event(GetCostOfCoffee, _) => sender ! (stateData.costOfCoffee); stay()
case Event(GetCostOfCoffee, machineData) => sender ! (machineData.costOfCoffee); stay()
用GetNumberOfCoffee和SetNumberOfCoffee設置最大咖啡數的實現幾乎與設置價格的方法差不多。我們先跳過這個來到更有趣的部分 - 買咖啡。
買咖啡
當咖啡愛好者為咖啡交了錢，我們還不能讓咖啡機做咖啡，要等到得到了一杯咖啡的錢才行。而且如果多給了現金，我們還要找零錢，所以，例子會變成這樣：

直到用戶開始存錢了，我們開始追蹤他的存款并stay在Open狀態。
2.當現金數達到一杯咖啡的錢了，我們會轉移成ReadyToBuy狀態并允許他買咖啡。

在ReadyToBuy狀態，他可以改變主意Cancel取消這次交易并拿到所有的退款Balance。

如果他想要喝咖啡，它發給咖啡機BrewCoffee煮咖啡的消息。（事實上，我們的代碼里并不會分發咖啡。我們只是從用戶的存款里減掉了咖啡的價格并找零。）

讓我們看下以下的用例
用例1 用戶存錢單但存的錢低于咖啡的價格
用例開始設置咖啡的價格為5元并且咖啡總數為10。 我們存2元并檢查機器是不是在Open狀態并且咖啡總數仍然是10.
it("should stay at Transacting when the Deposit is less then the price of the coffee") {
val coffeeMachine = TestActorRef(Props(new CoffeeMachine()))
coffeeMachine ! SetCostOfCoffee(5)
coffeeMachine ! SetNumberOfCoffee(10)
coffeeMachine ! SubscribeTransitionCallBack(testActor)
expectMsg(CurrentState(coffeeMachine, Open))
coffeeMachine ! Deposit(2)
coffeeMachine ! GetNumberOfCoffee
expectMsg(10)
}
我們怎樣確保機器在Open狀態？
每個FSM都能處理一條叫FSM.SubscribeTransitionCallBack(callerActorRef)的特殊消息，能讓調用者在任何狀態變動時被通知。第一條發給訂閱者的通知消息是CurrentState， 告訴我們FSM在哪個狀態。 這之后會有若干條Transition消息。
實現
我們繼續存錢并維持在Open狀態并等待存更多的錢
when(Open) {
...
...
case Event(Deposit(value), MachineData(currentTxTotal, costOfCoffee, coffeesLeft)) if (value + currentTxTotal) < stateData.costOfCoffee => {
val cumulativeValue = currentTxTotal + value
stay using stateData.copy(currentTxTotal = cumulativeValue)
}
用例2和4 - 用戶存錢并達到咖啡的價錢
測試用例1 - 存與咖啡價格等值的錢
我們的用例啟動機器，確認是否當前狀態是Open并存5元錢。 我們之后假定機器狀態從Open到ReadyToBuy，這可以通過接受一條Transition消息來證明咖啡機狀態的變更。在第一個例子，轉換是從Open到ReadyToBuy。
下一步我們讓凱飛機BrewCoffee煮咖啡，這時應該會有一條轉換，ReadToBuy到Open。 最終我們斷言咖啡機中的數量（就是9）。
it("should transition to ReadyToBuy and then Open when the Deposit is equal to the price of the coffee") {
val coffeeMachine = TestActorRef(Props(new CoffeeMachine()))
coffeeMachine ! SetCostOfCoffee(5)
coffeeMachine ! SetNumberOfCoffee(10)
coffeeMachine ! SubscribeTransitionCallBack(testActor)
expectMsg(CurrentState(coffeeMachine, Open))
coffeeMachine ! Deposit(5)
expectMsg(Transition(coffeeMachine, Open, ReadyToBuy))
coffeeMachine ! BrewCoffee
expectMsg(Transition(coffeeMachine, ReadyToBuy, Open))
coffeeMachine ! GetNumberOfCoffee
expectMsg(9)
}
測試用例2 - 存大于咖啡價格的錢
第二個例子跟第一個比有90%一樣，除了我們存在錢更多了（是6元）。 因為我們把咖啡價格設為5元， 現在我們期望應該有一塊錢的Balance找零消息
it("should transition to ReadyToBuy and then Open when the Deposit is greater than the price of the coffee") {
val coffeeMachine = TestActorRef(Props(new CoffeeMachine()))
coffeeMachine ! SetCostOfCoffee(5)
coffeeMachine ! SetNumberOfCoffee(10)
coffeeMachine ! SubscribeTransitionCallBack(testActor)
expectMsg(CurrentState(coffeeMachine, Open))
coffeeMachine ! Deposit(2)
coffeeMachine ! Deposit(2)
coffeeMachine ! Deposit(2)
expectMsg(Transition(coffeeMachine, Open, ReadyToBuy))
coffeeMachine ! BrewCoffee
expectMsgPF(){
case Balance(value)=>value==1
}
expectMsg(Transition(coffeeMachine, ReadyToBuy, Open))
coffeeMachine ! GetNumberOfCoffee
expectMsg(9)
}
實現
這個實現比之前的測試用例簡單。如果存款大于咖啡價格，那么我們轉到goto ReadyToBuy狀態。
when(Open){
...
...
case Event(Deposit(value), MachineData(currentTxTotal, costOfCoffee, coffeesLeft)) if (value + currentTxTotal) >= stateData.costOfCoffee => {
goto(ReadyToBuy) using stateData.copy(currentTxTotal = currentTxTotal + value)
}

一旦轉到ReadyToBuy狀態， 當用戶發送BrewCoffee，我們檢查是否有零錢找零。
when(ReadyToBuy) {
case Event(BrewCoffee, MachineData(currentTxTotal, costOfCoffee, coffeesLeft)) => {
val balanceToBeDispensed = currentTxTotal - costOfCoffee
logger.debug(s"Balance is $balanceToBeDispensed")
if (balanceToBeDispensed > 0) {
sender ! Balance(value = balanceToBeDispensed)
goto(Open) using stateData.copy(currentTxTotal = 0, coffeesLeft = coffeesLeft - 1)
}
else goto(Open) using stateData.copy(currentTxTotal = 0, coffeesLeft = coffeesLeft - 1)
}
}
用例3 用戶要取消交易
實際上， 用戶應該可以在交易的任何時間點Cancel取消，無論他在什么狀態。我們之前在第一部分討論過，最好的保存這里通用消息的地方在whenUnhandled代碼塊。我們要確定用戶在取消前是否存了一些錢，我們要還給他們。
實現
whenUnhandled {
...
...
case Event(Cancel, MachineData(currentTxTotal, _, _)) => {
sender ! Balance(value = currentTxTotal)
goto(Open) using stateData.copy(currentTxTotal = 0)
}
}
測試用例
這個例子跟我們以上看到的差不多，除了找零。
it("should transition to Open after flushing out all the deposit when the coffee is canceled") {
val coffeeMachine = TestActorRef(Props(new CoffeeMachine()))
coffeeMachine ! SetCostOfCoffee(5)
coffeeMachine ! SetNumberOfCoffee(10)
coffeeMachine ! SubscribeTransitionCallBack(testActor)
expectMsg(CurrentState(coffeeMachine, Open))
coffeeMachine ! Deposit(2)
coffeeMachine ! Deposit(2)
coffeeMachine ! Deposit(2)
expectMsg(Transition(coffeeMachine, Open, ReadyToBuy))
coffeeMachine ! Cancel
expectMsgPF(){
case Balance(value)=>value==6
}
expectMsg(Transition(coffeeMachine, ReadyToBuy, Open))
coffeeMachine ! GetNumberOfCoffee
expectMsg(10)
}

代碼
我不想煩死你所以跳過了解釋ShutDownMachine消息和PowerOff狀態，如果你想要解釋，可以留言。
像之前一樣，代碼在github

---

文章來自微信平臺「麥芽面包」
微信公眾號「darkjune_think」轉載請注明。
如果覺得有趣，微信掃一掃關注公眾號。

文章出處
微服務是大型分布式系統的基本組成部分，在面對海量用戶時在設計上考慮橫向擴展后，應用在集群間采用RPC調用，微服務的使用與目前互聯網服務的體量是分不開的，是控制復雜度的一種手段。
早期大家在處理http請求時由于單機linux的文件句柄數量上限，或單web容器的單機極限等原因開始采用服務器集群(老外叫的更貼切-服務器農場，含義既跟老農養雞養鴨一樣，養一堆服務器)，每個服務器處理一部分http請求；每個服務器的處理能力是不一樣的(新老機器混雜，虛擬化超賣，容器隔離性能)此時就要考慮如何讓機器物盡其用，能者多勞，不能者少勞，負載均衡設施那是必不可少的(無論是輪負載還是硬負載)，這里還只是http層的處理；
大家知道http就是個文本協議，傳來傳去對瀏覽器而言就是拿到服務器上傳回的文本，瀏覽器對遠方服務怎么生成這個并不關心。現在大多數網站已經都是動態網站了，比如淘寶首頁上的各種商品，每天都有上千萬的商家進行發布和編輯，我們看到的每個時間段都不一定是同樣的東西，都是動態的。而網頁中又會混雜很多靜態的東西， 例如頁面的js，圖片jpg/png等東西，從性能優化的角度講，這兩塊東西肯定是各用一套專門的服務來處理才是最優解， 業內對純http與后端動態服務的處理也有若干方案，如動態語言+fastCGI，Java方案有Servlet。 在此種方案的基礎上， 如果網站規模很大， 希望前端工程與后端工程的人解耦，走rest方式讓兩個團隊的人可以各干各的，只要最后做系統集成時才聯調一下也是個不錯的選擇。
以上只是http容器與一個宏觀后端服務器間的交互，那么當規模再大，后端服務之間也需要拆分了，每個服務做自己的事，控制一定量的復雜度， 比如讓一個業務開發再去想用戶的賬戶都怎么管理，異常怎么處理太浪費時間了，不利于快速沖業務。幾年前J2EE對此的答案就是SOA架構，那是還沒有走分布式架構體系，相關服務的處理還是考慮抽象出ESB總線，可以與計算機主板的總線對比，各個部件（組件）間的通信都通過總線（ESB）來處理。進入分布式系統的時代后，服務仍然在，只是變成了所謂的微服務了，從服務的角度并沒有什么不同。
至此，web app從進程內通信轉向進程間通信的趨勢已經塵埃落定了，通過增加服務器的方式進行橫向擴展也只能采用進程間通信的方式。剩下的就是如何讓工程師更快速方便的批量將應用的代碼發布上線，對于java應用來說，就是盡可能快的將war包（傳統，提供http）或jar包（springboot 服務式）復制到servlet容器（依賴容器啟動）或自包含自啟動的jar包（本質也是提供tcp服務端口連接）。 此處如果運維支撐環境做得不夠好，會有很多環境上的線上問題，因為開發的日常環境可能與線上環境不同，運維人員此時努力要做的就是維護日常與線上環境的基線， 不要有任何的不同， 避免出現詭異的問題。 這個問題現在有一種方式可以解，前景看起來還不錯，就是用docker化，讓開發保證自己打出的docker包是可運行的，由于docker容器與OS之間也做到了一層隔離，docker自己的生態環境內不會受到外部各種linux方言版本的影響，理論上講問題會更少，能做到拆箱即用。如果能大面積鋪開docker形式的運維部署系統，devops的時代也就到了，系統管理員只需要關注系統的安裝，對于黑箱的docker發布來說，想辦法快速在集群間分發下去，啟動即可。
不得不佩服美國同行的理論能力，在1985年就已經發布了論文解決大規模分布式系統的配置管理問題 Dynamic Configuration for Distributed Systems,

https://spiral.imperial.ac.uk/bitstream/10044/1/452/1/Dynamic%20Configuration%20for%20Distributed.pdf

那時候14K的貓都還沒有， BBS還在初級階段，就已經能YY到今天大規模分布式系統要解決的問題了。
題圖：福克斯RS
福克斯RS以普通福克斯為基礎研發出來的性能車型，但絕大部分的零部件都經過重新的設計。在外觀方面，它采用了全新的車身空氣動力學套件，使其不僅擁有了更為激進的外形，而且空氣動力學性能更加優異。
動力方面：全新福克斯RS搭載的是一臺2.3T渦輪增壓發動機，其最大輸出功率為350馬力，峰值扭矩為440牛·米，超增壓模式下可達到470牛·米。傳動系統匹配6速手動變速箱，并配備有四驅系統。官方稱，該車0-100km/h加速時間為4.7秒，最高車速為266km/h。

---

文章來自微信平臺「麥芽面包」
微信公眾號「darkjune_think」
轉載請注明。