スーパービジョンと耐障害性

レジリエントなシステムを構築する

親子スーパービジョン

障害への対応を決定する

OneForOne戦略

OneForOneの動作

import akka.actor._
import akka.actor.SupervisorStrategy._
import scala.concurrent.duration._

object OneForOneExample extends App {
  class Child extends Actor {
    override def preStart(): Unit = {
      println(s"Child ${self.path.name}: Started!")
    }
    override def postStop(): Unit = {
      println(s"Child ${self.path.name}: Stopped!")
    }
    override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
      println(s"Child ${self.path.name}: Pre-restart due to ${reason.getMessage}")
    }
    override def postRestart(reason: Throwable): Unit = {
      println(s"Child ${self.path.name}: Post-restart!")
    }
    def receive = {
      case "divideByZero" => throw new ArithmeticException("Oops, dividing by zero!")
      case "nullPointer"  => throw new NullPointerException("Something is null!")
      case msg            => println(s"Child ${self.path.name}: Received $msg")
    }
  }

  class Parent extends Actor {
    override val supervisorStrategy: SupervisorStrategy =
      OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1.minute) {
        case _: ArithmeticException => Restart
        case _: NullPointerException => Stop
        case _                      => Escalate
      }

    val child = context.actorOf(Props[Child], "myChild")

    def receive = {
      case msg => child ! msg
    }
  }

  val system = ActorSystem("OneForOneSystem")
  val parent = system.actorOf(Props[Parent], "myParent")

  parent ! "Hello"
  parent ! "divideByZero" // Should cause a restart
  Thread.sleep(100)
  parent ! "Hello again"
  Thread.sleep(1000) // Give time for restart to complete

  parent ! "nullPointer"  // Should cause a stop
  Thread.sleep(100)
  parent ! "Hello after null" // This message will not be processed by child
  Thread.sleep(2000)

  system.terminate()
}

OneForOneの動作を確認する

AllForOne戦略

AllForOneの動作

import akka.actor._
import akka.actor.SupervisorStrategy._
import scala.concurrent.duration._

object AllForOneExample extends App {
  class Child extends Actor {
    override def preStart(): Unit = {
      println(s"Child ${self.path.name}: Started!")
    }
    override def postStop(): Unit = {
      println(s"Child ${self.path.name}: Stopped!")
    }
    override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
      println(s"Child ${self.path.name}: Pre-restart due to ${reason.getMessage}")
    }
    override def postRestart(reason: Throwable): Unit = {
      println(s"Child ${self.path.name}: Post-restart!")
    }
    def receive = {
      case "fail" => throw new RuntimeException("Child failed!")
      case msg    => println(s"Child ${self.path.name}: Received $msg")
    }
  }

  class Parent extends Actor {
    override val supervisorStrategy: SupervisorStrategy =
      AllForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1.minute) {
        case _: RuntimeException => Restart
        case _                   => Escalate
      }

    val childA = context.actorOf(Props[Child], "childA")
    val childB = context.actorOf(Props[Child], "childB")

    def receive = {
      case "failA" => childA ! "fail"
      case "msgB"  => childB ! "Hello from B"
      case msg     => println(s"Parent received: $msg")
    }
  }

  val system = ActorSystem("AllForOneSystem")
  val parent = system.actorOf(Props[Parent], "myParent")

  parent ! "msgB" // Child B receives a message
  Thread.sleep(100)

  parent ! "failA" // Child A fails, causing all children to restart
  Thread.sleep(1000)

  parent ! "msgB" // Child B (restarted) receives another message
  Thread.sleep(2000)

  system.terminate()
}

判定ロジックをカスタマイズする

高度なスーパービジョン

import akka.actor._
import akka.actor.SupervisorStrategy._
import scala.concurrent.duration._

object CustomDeciderExample extends App {
  class Worker extends Actor {
    def receive = {
      case "criticalError" => throw new IllegalStateException("Critical state!")
      case "minorError"    => throw new IllegalArgumentException("Bad input!")
      case msg             => println(s"Worker: Received $msg")
    }
  }

  class Supervisor extends Actor {
    override val supervisorStrategy: SupervisorStrategy =
      OneForOneStrategy(maxNrOfRetries = 3, withinTimeRange = 10.seconds) {
        case _: IllegalArgumentException => Resume // Minor error, just resume
        case _: IllegalStateException    => Restart // Critical, restart
        case _: Exception                => Stop    // Unknown, stop
        case _                           => Escalate // Others, escalate
      }

    val worker = context.actorOf(Props[Worker], "myWorker")

    def receive = {
      case msg => worker ! msg
    }
  }

  val system = ActorSystem("CustomDeciderSystem")
  val supervisor = system.actorOf(Props[Supervisor], "mySupervisor")

  supervisor ! "Hello"
  supervisor ! "minorError" // Should Resume
  Thread.sleep(100)
  supervisor ! "Hello again" // Worker should still be active
  Thread.sleep(500)

  supervisor ! "criticalError" // Should Restart
  Thread.sleep(100)
  supervisor ! "After critical" // Worker should be restarted
  Thread.sleep(1500)

  system.terminate()
}

理解度を確認する

振り返り：フォールトトレランス