Akka 2.4.0がリリースされたのでAkka Persistenceを試してみる。

　こんにちは。先月、待望のAkka 2.4.0がリリースされたので、 2.4.0の目玉機能であるAkka Persistenceを試してみました。
　いやー、待ちました。なかなかexperimental(実験的な機能の意味)が外れないのでやきもきしていましたが、やっとakka-persistence-experimentalからakka-persistenceになったので、自信を持ってブログで紹介することが出来るようになりました。リリースお疲れ様です！

Akka 2.4.0についてはこちら↓
Akka 2.4.0 Released! | Akka

Akka Persistenceについてはこちら↓
Persistence — Akka Documentation

今回の記事は、Akkaは知っているけどAkka Persistenceは知らないよという方向けの記事です。
ん？Elixir？知らない子ですね。

Akka Persistenceとは

　Akka Persistenceを使うとActorの内部状態を永続化することが出来るようになります。マシンやJVMがシャットダウンしても、前回のActorの状態を引き継いだActorを立ち上げることが出来ます。
Akka Persistenceはイベントソーシングの考え方に基づき、Actorの内部状態を直接保存するのではなく、そのActorに起こった一連のイベントを保存し、Actorの復元の際にそのイベントを順番に再生することで内部状態を復元します。

イベントソーシングを使うことによって、

履歴管理が不要
溜まったイベントをデータ解析に使える
イベントは基本的には追記のみなのでパフォーマンスが出やすい
ActorのID(persistenceId)に基づいて水平分割できるのでスケールしやすい
ORMが不要
スキーマの概念がないので、イベントの種類を増やすことでモデルを簡単に拡張できる。
と、様々な特徴があります。

　イベントソーシングはしばしば、コマンドクエリ責務分離(CQRS: Command Query Responsibility Segregation)と併用されて、コマンドによって引き起こされたドメインイベントをイベントソーシングで保存して（コマンドサイド）、 Actorの内部状態を好きなように、DBに投影して、クエリに応答する（クエリサイド）。といった使われ方をされます。

　今回は、イベントソーシングやCQRSについては紹介しきれませんが、 Akka Persistenceの簡単な紹介から、イベントソーシングやCQRSについて興味を持って頂けたら幸いです。

はじめての永続化

　全ソースはこちら

　Akka Persistenceを使うためには libraryDependenciesに
"com.typesafe.akka" %% "akka-persistence" % "2.4.0" を追加します。

例として、単純に0から1ずつカウントアップして、カウントした値を保持するActorを作ってみます。

class CountUpActor extends PersistentActor with ActorLogging {

  var count: Int = 0

  override def persistenceId: String = self.path.name

  override def receiveCommand: Receive = {
    case c: CountUp =>
      log.info("receive command {}", c)
      persist(Increased(c.count)) {
        event =>
          count += event.diff
          log.info("current count {}", count)
      }
  }

  override def receiveRecover: Receive = {
    case e: Increased =>
      log.info("receive recover {}", e)
      count += e.diff
  }
}

object CountUpActor {
  def props = Props[CountUpActor]

  case class CountUp(count: Int)

  case class Increased(diff: Int)

}

　PersistentActorを継承しています。
def receiveCommand で、CountUpコマンドを受け取ってpersist(Increased(c.count)) で永続化しています。
ポイントは永続化しているのは、内部状態(count)ではなく、増加した(Increased)というイベントを永続化しているところです。また、永続化した後に count += event.diff と内部状態を更新しているので、状態は変わったのに、永続化されていないといったことは起こりません。

永続化したイベントは次回Actorの起動時に、 def receiveRecover でハンドルできます。ここでは復元したイベントをもとにcountを増加させて、前回終了時のcountの値を復元しています。

実行

　さっそく動かしてみましょう。Mainクラスはこんな感じにCountUpコマンドを3回送っています。

object ApplicationMain extends App {
  val system = ActorSystem("MyActorSystem")

  val actor = system.actorOf(CountUpActor.props, "c1")

  actor ! CountUp(1)
  actor ! CountUp(1)
  actor ! CountUp(1)

  Await.result(system.whenTerminated, Duration.Inf)
}

一回目の実行結果

[info] Running com.example.ApplicationMain
[info][akka://MyActorSystem/user/c1] receive command CountUp(1)
[info][akka://MyActorSystem/user/c1] current count 1
[info][akka://MyActorSystem/user/c1] receive command CountUp(1)
[info][akka://MyActorSystem/user/c1] current count 2
[info][akka://MyActorSystem/user/c1] receive command CountUp(1)
[info][akka://MyActorSystem/user/c1] current count 3

二回目の実行結果

[info] Running com.example.ApplicationMain
[info][akka://MyActorSystem/user/c1] receive recover Increased(1)
[info][akka://MyActorSystem/user/c1] receive recover Increased(1)
[info][akka://MyActorSystem/user/c1] receive recover Increased(1)
[info][akka://MyActorSystem/user/c1] receive command CountUp(1)
[info][akka://MyActorSystem/user/c1] current count 4
[info][akka://MyActorSystem/user/c1] receive command CountUp(1)
[info][akka://MyActorSystem/user/c1] current count 5
[info][akka://MyActorSystem/user/c1] receive command CountUp(1)
[info][akka://MyActorSystem/user/c1] current count 6

　一回目の実行で countが3まで増加しています。
二回目の実行で receive recover Increased(1) が3回続き、1回目の実行で保存したイベントが復元されているのがわかります。その後、CoutUpコマンドを3回受け取り、正しく、合計6までカウントすることができています。

Persistence Plugin

　今回の例ではイベントの保存先はLevelDBを使っていますが、LevelDB以外にも使うことができて、自分で開発したプラグインや他の方が開発したプラグインをつかって様々なDB(永続化機構)にイベントを保存することができます。 http://akka.io/community/#plugins-to-akka-persistence
　テスト時は標準で準備されているインメモリプラグインを使うと良いと思います。
設定はapplication.confに書きます。
akka.persistence.journal.plugin="akka.persistence.journal.inmem"