摘要：源碼鏈接進(jìn)階持久化插件有同學(xué)可能會(huì)問(wèn)，我對(duì)不是很熟悉亦或者覺(jué)得單機(jī)存儲(chǔ)并不是安全，有沒(méi)有支持分布式數(shù)據(jù)存儲(chǔ)的插件呢，比如某爸的云數(shù)據(jù)庫(kù)答案當(dāng)然是有咯，良心的我當(dāng)然是幫你們都找好咯。

這次把這部分內(nèi)容提到現(xiàn)在寫(xiě)，是因?yàn)檫@段時(shí)間開(kāi)發(fā)的項(xiàng)目剛好在這一塊遇到了一些難點(diǎn)，所以準(zhǔn)備把經(jīng)驗(yàn)分享給大家，我們?cè)谑褂肁kka時(shí)，會(huì)經(jīng)常遇到一些存儲(chǔ)Actor內(nèi)部狀態(tài)的場(chǎng)景，在系統(tǒng)正常運(yùn)行的情況下，我們不需要擔(dān)心什么，但是當(dāng)系統(tǒng)出錯(cuò)，比如Actor錯(cuò)誤需要重啟，或者內(nèi)存溢出，亦或者整個(gè)系統(tǒng)崩潰，如果我們不采取一定的方案的話，在系統(tǒng)重啟時(shí)Actor的狀態(tài)就會(huì)丟失，這會(huì)導(dǎo)致我們丟失一些關(guān)鍵的數(shù)據(jù)，造成系統(tǒng)數(shù)據(jù)不一致的問(wèn)題。Akka作為一款成熟的生產(chǎn)環(huán)境應(yīng)用，為我們提供了相應(yīng)的解決方案就是Akka persistence。

萬(wàn)變不離其宗，數(shù)據(jù)的一致性是永恒的主題，一個(gè)性能再好的系統(tǒng)，不能保證數(shù)據(jù)的正確，也稱不上是一個(gè)好的系統(tǒng)，一個(gè)系統(tǒng)在運(yùn)行的時(shí)候難免會(huì)出錯(cuò)，如何保證系統(tǒng)在出錯(cuò)后能正確的恢復(fù)數(shù)據(jù)，不讓數(shù)據(jù)出現(xiàn)混亂是一個(gè)難題。使用Actor模型的時(shí)候，我們會(huì)有這么一個(gè)想法，就是能不對(duì)數(shù)據(jù)庫(kù)操作就盡量不對(duì)數(shù)據(jù)庫(kù)操作（這里我們假定我們的數(shù)據(jù)庫(kù)是安全，可靠的，能保證數(shù)據(jù)的正確性和一致性，比如使用國(guó)內(nèi)某云的云數(shù)據(jù)庫(kù)），一方面如果大量的數(shù)據(jù)操作會(huì)使數(shù)據(jù)庫(kù)面臨的巨大的壓力，導(dǎo)致崩潰，另一方面即使數(shù)據(jù)庫(kù)能處理的過(guò)來(lái)，比如一些count，update的大表操作也會(huì)消耗很多的時(shí)間，遠(yuǎn)沒(méi)有內(nèi)存中直接操作來(lái)的快，大大影響性能。但是又有人說(shuō)幾人內(nèi)存操作這么快，為什么不把數(shù)據(jù)都放內(nèi)存中呢？答案顯而易見(jiàn)，當(dāng)出現(xiàn)機(jī)器死機(jī)，或者內(nèi)存溢出等問(wèn)題時(shí)，數(shù)據(jù)很有可能就丟失了導(dǎo)致無(wú)法恢復(fù)。在這種背景下，我們是不是有一種比較好的解決方案，既能滿足需求又能用最小的性能消耗，答案就是上面我們的說(shuō)的Akka persistence。

在具體深入Akka persistence之前，我們可以先了解一下它的核心設(shè)計(jì)理念，其實(shí)簡(jiǎn)單來(lái)說(shuō)，我們可以利用一些thing來(lái)恢復(fù)Actor的狀態(tài)，這里的thing可以是日志、數(shù)據(jù)庫(kù)中的數(shù)據(jù)，亦或者是文件，所以說(shuō)它的本質(zhì)非常容易理解，在Actor處理的時(shí)候我們會(huì)保存一些數(shù)據(jù)，Actor在恢復(fù)的時(shí)候能根據(jù)這些數(shù)據(jù)恢復(fù)其自身的狀態(tài)。

所以Akka persistence 有以下幾個(gè)關(guān)鍵部分組成：

PersistentActor：任何一個(gè)需要持久化的Actor都必須繼承它，并必須定義或者實(shí)現(xiàn)其中的三個(gè)關(guān)鍵屬性：

 def persistenceId = "example" //作為持久化Actor的唯一表示，用于持久化或者查詢時(shí)使用

 def receiveCommand: Receive = ??? //Actor正常運(yùn)行時(shí)處理處理消息邏輯，可在這部分內(nèi)容里持久化自己想要的消息

 def receiveRecover: Receive = ??? //Actor重啟恢復(fù)是執(zhí)行的邏輯

相比普通的Actor，除receiveCommand相似以外，還必須實(shí)現(xiàn)另外兩個(gè)屬性。
另外在持久化Actor中還有另外兩個(gè)關(guān)鍵的的概念就是Journal和Snapshot，前者用于持久化事件，后者用于保存Actor的快照，兩者在Actor恢復(fù)狀態(tài)的時(shí)候都起到了至關(guān)重要的作用。

這里我首先會(huì)用一個(gè)demo讓大家能對(duì)Akka persistence的使用有一定了解的，并能大致明白它的工作原理，后面再繼續(xù)講解一些實(shí)戰(zhàn)可能會(huì)遇到的問(wèn)題。

假定現(xiàn)在有這么一個(gè)場(chǎng)景，現(xiàn)在假設(shè)有一個(gè)1w元的大紅包，瞬間可能會(huì)很多人同時(shí)來(lái)?yè)專總€(gè)人搶的金額也可能不一樣，場(chǎng)景很簡(jiǎn)單，實(shí)現(xiàn)方式也有很多種，但前提是保證數(shù)據(jù)的正確性，比如最普通的使用數(shù)據(jù)庫(kù)保證，但對(duì)這方面有所了解的同學(xué)都知道這并不是一個(gè)很好的方案，因?yàn)樾枰i，并需要大量的數(shù)據(jù)庫(kù)操作，導(dǎo)致性能不高，那么我們是否可以用Actor來(lái)實(shí)現(xiàn)這個(gè)需求么？答案是當(dāng)然可以。

我們首先來(lái)定義一個(gè)抽獎(jiǎng)命令，

case class LotteryCmd(
  userId: Long, // 參與用戶Id
  username: String, //參與用戶名
  email: String // 參與用戶郵箱
)

然后我們實(shí)現(xiàn)一個(gè)抽獎(jiǎng)Actor，并繼承PersistentActor作出相應(yīng)的實(shí)現(xiàn)：

case class LuckyEvent(  //抽獎(jiǎng)成功事件
    userId: Long,
    luckyMoney: Int
)
case class FailureEvent(  //抽獎(jiǎng)失敗事件
    userId: Long,
    reason: String
)
case class Lottery(
    totalAmount: Int,  //紅包總金額
    remainAmount: Int  //剩余紅包金額
) {
  def update(luckyMoney: Int) = {
    copy(
      remainAmount = remainAmount - luckyMoney
    )
  }
}
class LotteryActor(initState: Lottery) extends PersistentActor with ActorLogging{
  override def persistenceId: String = "lottery-actor-1"

  var state = initState  //初始化Actor的狀態(tài)

  override def receiveRecover: Receive = {
    case event: LuckyEvent =>
      updateState(event)  //恢復(fù)Actor時(shí)根據(jù)持久化的事件恢復(fù)Actor狀態(tài)
    case SnapshotOffer(_, snapshot: Lottery) =>
      log.info(s"Recover actor state from snapshot and the snapshot is ${snapshot}")
      state = snapshot //利用快照恢復(fù)Actor的狀態(tài)
    case RecoveryCompleted => log.info("the actor recover completed")
  }

  def updateState(le: LuckyEvent) =
    state = state.update(le.luckyMoney)  //更新自身狀態(tài)

  override def receiveCommand: Receive = {
    case lc: LotteryCmd =>
      doLottery(lc) match {     //進(jìn)行抽獎(jiǎng)，并得到抽獎(jiǎng)結(jié)果，根據(jù)結(jié)果做出不同的處理
        case le: LuckyEvent =>  //抽到隨機(jī)紅包
          persist(le) { event =>
            updateState(event)
            increaseEvtCountAndSnapshot()
            sender() ! event
          }
        case fe: FailureEvent =>  //紅包已經(jīng)抽完
          sender() ! fe
      }
    case "saveSnapshot" =>  // 接收存儲(chǔ)快照命令執(zhí)行存儲(chǔ)快照操作
      saveSnapshot(state)
    case SaveSnapshotSuccess(metadata) =>  ???  //你可以在快照存儲(chǔ)成功后做一些操作，比如刪除之前的快照等
  }

  private def increaseEvtCountAndSnapshot() = {
    val snapShotInterval = 5
    if (lastSequenceNr % snapShotInterval == 0 && lastSequenceNr != 0) {  //當(dāng)有持久化5個(gè)事件后我們便存儲(chǔ)一次當(dāng)前Actor狀態(tài)的快照
      self ! "saveSnapshot"
    }
  }

  def doLottery(lc: LotteryCmd) = {  //抽獎(jiǎng)邏輯具體實(shí)現(xiàn)
    if (state.remainAmount > 0) {
      val luckyMoney = scala.util.Random.nextInt(state.remainAmount) + 1
      LuckyEvent(lc.userId, luckyMoney)
    }
    else {
      FailureEvent(lc.userId, "下次早點(diǎn)來(lái)，紅包已被抽完咯！")
    }
  }
}

程序很簡(jiǎn)單，關(guān)鍵位置我也給了注釋，相信大家對(duì)Actor有所了解的話很容易理解，當(dāng)然要是有些疑惑，可以看看我之前寫(xiě)的文章，下面我們就對(duì)剛才寫(xiě)的抽紅包Actor進(jìn)行測(cè)試：

object PersistenceTest extends App {
  val lottery = Lottery(10000,10000)
  val system = ActorSystem("example-05")
  val lotteryActor = system.actorOf(Props(new LotteryActor(lottery)), "LotteryActor-1")  //創(chuàng)建抽獎(jiǎng)Actor
  val pool: ExecutorService = Executors.newFixedThreadPool(10)
  val r = (1 to 100).map(i =>
    new LotteryRun(lotteryActor, LotteryCmd(i.toLong,"godpan","xx@gmail.com"))  //創(chuàng)建100個(gè)抽獎(jiǎng)?wù)埱?  )
  r.map(pool.execute(_))  //使用線程池來(lái)發(fā)起抽獎(jiǎng)?wù)埱?，模擬同時(shí)多人參加
  Thread.sleep(5000)
  pool.shutdown()
  system.terminate()
}

class LotteryRun(lotteryActor: ActorRef, lotteryCmd: LotteryCmd) extends Runnable { //抽獎(jiǎng)?wù)埱?  implicit val timeout = Timeout(3.seconds)
  def run: Unit = {
    for {
      fut <- lotteryActor ? lotteryCmd
    } yield fut match {  //根據(jù)不同事件顯示不同的抽獎(jiǎng)結(jié)果
      case le: LuckyEvent => println(s"恭喜用戶${le.userId}抽到了${le.luckyMoney}元紅包")
      case fe: FailureEvent =>  println(fe.reason)
      case _ => println("系統(tǒng)錯(cuò)誤，請(qǐng)重新抽取")
    }
  }
}

運(yùn)行程序,我們可能看到以下的結(jié)果：

下面我會(huì)把persistence actor在整個(gè)運(yùn)行過(guò)程的步驟給出，幫助大家理解它的原理：

1.初始化Persistence Actor

1.1若是第一次初始化，則與正常的Actor的初始化一致。

1.2若是重啟恢復(fù)Actor，這根據(jù)Actor之前持久的數(shù)據(jù)恢復(fù)。

1.2.1從快照恢復(fù)，可快速恢復(fù)Actor，但并非每次持久化事件都會(huì)保存快照，在快照完整的情況下，Actor優(yōu)先從快照恢復(fù)自身狀態(tài)。

1.2.2從事件（日志，數(shù)據(jù)庫(kù)記錄等）恢復(fù)，通過(guò)重放持久化事件恢復(fù)Actor狀態(tài)，比較關(guān)鍵。

2.接收命令進(jìn)行處理，轉(zhuǎn)化為需要持久化的事件（持久化的事件盡量只包含關(guān)鍵性的數(shù)據(jù)）使用Persistence Actor的持久化方法進(jìn)行持久化（上述例子中的persist，后面我會(huì)講一下批量持久化），并處理持久化成功后的邏輯處理，比如修改Actor狀態(tài)，向外部Actor發(fā)送消息等。

3.若是我們需要存儲(chǔ)快照，那么可以主動(dòng)指定存儲(chǔ)快照的頻率，比如持久化事件100次我們就存儲(chǔ)一次快照，這個(gè)頻率應(yīng)該要考慮實(shí)際的業(yè)務(wù)場(chǎng)景，在存儲(chǔ)快照成功后我們也可以執(zhí)行一些操作。

總的來(lái)說(shuō)Persistence Actor運(yùn)行時(shí)的大致操作就是以上這些，當(dāng)然它是r如何持久化事件，恢復(fù)時(shí)的機(jī)制是怎么樣的等有興趣的可以看一下Akka源碼。

首先我們必須加載相應(yīng)的依賴包，在bulid.sbt中加入以下依賴：

libraryDependencies ++= Seq(
"com.typesafe.akka" %% "akka-actor" % "2.4.16",  //Akka actor 核心依賴
  "com.typesafe.akka" %% "akka-persistence" % "2.4.16", //Akka persistence 依賴
  "org.iq80.leveldb"            % "leveldb"          % "0.7", //leveldb java版本依賴
  "org.fusesource.leveldbjni"   % "leveldbjni-all"   % "1.8", //leveldb java版本依賴
  "com.twitter"              %% "chill-akka"                  % "0.8.0" //事件序列化依賴
)

另外我們還需在application.conf加入以下配置:

akka.persistence.journal.plugin = "akka.persistence.journal.leveldb"
akka.persistence.snapshot-store.plugin = "akka.persistence.snapshot-store.local"

akka.persistence.journal.leveldb.dir = "log/journal"
akka.persistence.snapshot-store.local.dir = "log/snapshots"

# DO NOT USE THIS IN PRODUCTION !!!
# See also https://github.com/typesafehub/activator/issues/287
akka.persistence.journal.leveldb.native = false  //因?yàn)槲覀儽镜夭](méi)有安裝leveldb，所以這個(gè)屬性置為false，但是生產(chǎn)環(huán)境并不推薦使用

akka.actor.serializers {
  kryo = "com.twitter.chill.akka.AkkaSerializer"
}

akka.actor.serialization-bindings {
  "scala.Product" = kryo
  "akka.persistence.PersistentRepr" = kryo
}

至此為止我們整個(gè)Akka persistence demo已經(jīng)搭建好了，可以正常運(yùn)行了，有興趣的同學(xué)可以下載源碼。源碼鏈接

有同學(xué)可能會(huì)問(wèn)，我對(duì)leveldb不是很熟悉亦或者覺(jué)得單機(jī)存儲(chǔ)并不是安全，有沒(méi)有支持分布式數(shù)據(jù)存儲(chǔ)的插件呢，比如某爸的云數(shù)據(jù)庫(kù)？答案當(dāng)然是有咯，良心的我當(dāng)然是幫你們都找好咯。

1.akka-persistence-sql-async: 支持MySQL和PostgreSQL，另外使用了全異步的數(shù)據(jù)庫(kù)驅(qū)動(dòng)，提供異步非阻塞的API，我司用的就是它的變種版，6的飛起。項(xiàng)目地址

2.akka-persistence-cassandra: 官方推薦的插件，使用寫(xiě)性能very very very fast的cassandra數(shù)據(jù)庫(kù)，是幾個(gè)插件中比較流行的一個(gè)，另外它還支持persistence query。項(xiàng)目地址

3.akka-persistence-redis: redis應(yīng)該也很符合Akka persistence的場(chǎng)景，熟悉redis的同學(xué)可以使用看看。項(xiàng)目地址

4.akka-persistence-jdbc: 怎么能少了jdbc呢？不然怎么對(duì)的起java爸爸呢，支持scala和java哦。項(xiàng)目地址

相應(yīng)的插件的具體使用可以看該項(xiàng)目的具體介紹使用，我看了下相對(duì)來(lái)說(shuō)都是比較容易的。

上面說(shuō)到我司用的是akka-persistence-sql-async插件，所以我們是將事件和快照持久化到數(shù)據(jù)庫(kù)的，一開(kāi)始我也是像上面demo一樣，每次事件都會(huì)持久化到數(shù)據(jù)庫(kù)，但是后來(lái)在性能測(cè)試的時(shí)候，因?yàn)楸旧順I(yè)務(wù)場(chǎng)景對(duì)數(shù)據(jù)庫(kù)的壓力也比較大，在當(dāng)數(shù)據(jù)庫(kù)到達(dá)每秒1000+的讀寫(xiě)量后，另外說(shuō)明一下使用的是某云數(shù)據(jù)庫(kù)，性能中配以上，發(fā)現(xiàn)每次持久化的時(shí)間將近要15ms，這樣換算一下的話Actor每秒只能處理60~70個(gè)需要持久化的事件，而實(shí)際業(yè)務(wù)場(chǎng)景要求Actor必須在3秒內(nèi)返回處理結(jié)果，這種情況下導(dǎo)致大量消息處理超時(shí)得不到反饋，另外還有大量的消息得不到處理，導(dǎo)致系統(tǒng)錯(cuò)誤暴增，用戶體驗(yàn)下降，既然我們發(fā)現(xiàn)了問(wèn)題，那么我們能不能進(jìn)行優(yōu)化呢?事實(shí)上當(dāng)然是可以，既然單個(gè)插入慢，那么我們能不能批量插入呢，Akka persistence為我們提供了persistAll方法，下面我就對(duì)上面的demo進(jìn)行一下改造，讓其變成批量持久化：

class LotteryActorN(initState: Lottery) extends PersistentActor with ActorLogging{
  override def persistenceId: String = "lottery-actor-2"

  var state = initState  //初始化Actor的狀態(tài)

  override def receiveRecover: Receive = {
    case event: LuckyEvent =>
      updateState(event)  //恢復(fù)Actor時(shí)根據(jù)持久化的事件恢復(fù)Actor狀態(tài)
    case SnapshotOffer(_, snapshot: Lottery) =>
      log.info(s"Recover actor state from snapshot and the snapshot is ${snapshot}")
      state = snapshot //利用快照恢復(fù)Actor的狀態(tài)
    case RecoveryCompleted => log.info("the actor recover completed")
  }

  def updateState(le: LuckyEvent) =
    state = state.update(le.luckyMoney)  //更新自身狀態(tài)

  var lotteryQueue : ArrayBuffer[(LotteryCmd, ActorRef)] = ArrayBuffer()

  context.system.scheduler  //定時(shí)器，定時(shí)觸發(fā)抽獎(jiǎng)邏輯
    .schedule(
      0.milliseconds,
      100.milliseconds,
      new Runnable {
        def run = {
          self ! "doLottery"
        }
      }
    )

  override def receiveCommand: Receive = {
    case lc: LotteryCmd =>
      lotteryQueue = lotteryQueue :+ (lc, sender())  //參與信息加入抽獎(jiǎng)隊(duì)列
      println(s"the lotteryQueue size is ${lotteryQueue.size}")
      if (lotteryQueue.size > 5)  //當(dāng)參與人數(shù)有5個(gè)時(shí)觸發(fā)抽獎(jiǎng)
        joinN(lotteryQueue)
    case "doLottery" =>
      if (lotteryQueue.size > 0)
        joinN(lotteryQueue)
    case "saveSnapshot" =>  // 接收存儲(chǔ)快照命令執(zhí)行存儲(chǔ)快照操作
      saveSnapshot(state)
    case SaveSnapshotSuccess(metadata) =>  ???  //你可以在快照存儲(chǔ)成功后做一些操作，比如刪除之前的快照等
  }

  private def joinN(lotteryQueue: ArrayBuffer[(LotteryCmd, ActorRef)]) = {  //批量處理抽獎(jiǎng)結(jié)果
    val rs = doLotteryN(lotteryQueue)
    val success = rs.collect {  //得到其中中獎(jiǎng)的相應(yīng)信息
      case (event: LuckyEvent, ref: ActorRef) =>
        event -> ref
    }.toMap
    val failure = rs.collect {  //得到其中未中獎(jiǎng)的相應(yīng)信息
      case (event: FailureEvent, ref: ActorRef) => event -> ref
    }
    persistAll(success.keys.toIndexedSeq) {  //批量持久化中獎(jiǎng)用戶事件
      case event =>  println(event)
        updateState(event)
        increaseEvtCountAndSnapshot()
        success(event) ! event
    }
    failure.foreach {
      case (event, ref) => ref ! event
    }
    this.lotteryQueue.clear()  //清空參與隊(duì)列
  }


  private def increaseEvtCountAndSnapshot() = {
    val snapShotInterval = 5
    if (lastSequenceNr % snapShotInterval == 0 && lastSequenceNr != 0) {  //當(dāng)有持久化5個(gè)事件后我們便存儲(chǔ)一次當(dāng)前Actor狀態(tài)的快照
      self ! "saveSnapshot"
    }
  }

  private def doLotteryN(lotteryQueue: ArrayBuffer[(LotteryCmd, ActorRef)]) = {  //抽獎(jiǎng)邏輯具體實(shí)現(xiàn)
    var remainAmount = state.remainAmount
    lotteryQueue.map(lq =>
      if (remainAmount > 0) {
        val luckyMoney = scala.util.Random.nextInt(remainAmount) + 1
        remainAmount = remainAmount - luckyMoney
        (LuckyEvent(lq._1.userId, luckyMoney),lq._2)
      }
      else {
        (FailureEvent(lq._1.userId, "下次早點(diǎn)來(lái)，紅包已被抽完咯！"),lq._2)
      }
    )
  }
}

這是改造后的參與Actor，實(shí)現(xiàn)了批量持久的功能，當(dāng)然這里為了給發(fā)送者返回消息，處理邏輯稍微復(fù)雜了一點(diǎn)，不過(guò)真實(shí)場(chǎng)景可能會(huì)更復(fù)雜，相關(guān)源碼也在剛才的項(xiàng)目上。

另外Akka Persistence還提供了Query接口，用于需要查詢持久化事件的需求，這部分內(nèi)容可能要根據(jù)實(shí)際業(yè)務(wù)場(chǎng)景考慮是否需要應(yīng)用，我就不展開(kāi)講了，另外我也寫(xiě)了一個(gè)小demo在項(xiàng)目中，想要嘗試的同學(xué)也可以試試。