commons-daemonとAkkaでバッチアプリケーションの作成

ScalaとApache Commons Daemonで常駐プログラムを作るを参考に、akka-actorを使う場合はどうするかサンプルを作ってみる。

ファイル構成は以下の様なベタ書きを想定。パッケージを分ける場合は適宜importを補完してください。

Main.scala
ApplicationDaemon.scala
ReaperActor.scala
SampleActor.scala

Main

実行起点。これ自体はすぐに終了。

Daemonの方でスレッドの終了を待機しているので、ActorSystem#awaitTerminationとかは必要ありません。

Main.scala

object Main extends App {
  new ApplicationDaemon("sample").start
}

ApplicationDaemon

Daemonインターフェースの実装を行う。外部から停止指示があった場合、stop()が呼ばれたりする。

ちなみに、Actorのコンパニオンオブジェクトでpropsを実装するのが好み。引数を定義したりしやすいので。

追記：好みというか、推奨されるパターンですね。http://doc.akka.io/docs/akka/current/scala/actors.html#Recommended_Practices

今回はstart(), stop()のみ処理を記述したが、他のメソッドも利用を検討して欲しい。

ApplicaitonDaemon.scala

import org.apache.commons.daemon.{Daemon, DaemonContext}

class ApplicationDaemon(appName: String) extends Daemon {

  private[this] val actorSystem = ActorSystem(appName)

  override def init(context: DaemonContext) {}
  override def destroy {}

  override def start {
    val reaper = actorSystem.actorOf(ReaperActor.props, "reaper")

    val sampleActor = actorSystem.actorOf(SampleActor.props, "sample")
    reaper ! ReaperActor.Watch(sampleActor)

    sampleActor ! SampleActor.CountDown(10)
  }

  override def stop {
    if (! actorSystem.isTerminated) {
      // 通常はActorに終了指示を送るなど…
      actorSystem.shutdown
    }
  }
}

ReaperActor

死神。死亡を監視したいクリティカルなActorRefを送信しておいて、どれかが死亡した時に全てのアクターに終了指示を送ってActorSystemの終了に向かわせる。

全体的な流れの図や他の終了パターンは Let it crash • Shutdown Patterns in Akka 2 を参照すると良いです。

ReaperActor.scala

import akka.actor.{Actor,ActorRef,Props,Terminated,PoisonPill}

class ReaperActor extends Actor {
  private[this] val watchingActors = scala.collection.mutable.ArrayBuffer.empty[ActorRef]

  def receive {
    case ReaperActor.Watch(actor) =>
      context.watch(actor)
      watchingActors += actor
    case Terminated(actor) =>
      watchingActors -= actor
      if (watchingActors.isEmpty) { context.system.shutdown }
      // 通常は終了を知らせるメッセージを自前で定義して利用する
      watchingActors.foreach(_ ! PoisonPill)
  }
}

object ReaperActor {
  def props = Props[ReaperActor]
  case class Watch(actor: ActorRef)
}

SampleActor

サンプルの処理。自分自身の終了だけを管理する。
ここでは1秒おきにカウントダウンしていくアクターを作ってみる。

SampleActor.scala

import akka.actor.{Actor,ActorRef,Props}

class SampleActor extends Actor {

  import SampleActor._

  def receive {
    case CountDown(0) => 
      context.stop(self)
    case CountDown(n) =>
      println(n)
      Thread.sleep(1000)
      self ! CountDown(n-1)
  }
}

object SampleActor {
  def props = Props[SampleActor]
  case class CountDown(n: Int) { require(n >= 0) }
}