Akka（2）：Actor生命周期管理 - 监控和监视

import akka.actor._
object MyActor { //在这个伴生对象里申明MyActor所支持的功能指令
  sealed trait ActorCommands
  case object RunFuncA extends ActorCommands
  case object RunFuncB extends ActorCommands
}
//假设有funcA,funcB. 它们可以从任何JVM函数库里调用
val funcA : () => Any = ???
val funcB : () => Any = ???
class MyActor extends Actor {
  import MyActor._
  var stateValue: Any = _    //内部状态，代表这个Actor的当前运算结果
  override def receive: Receive = {
    case RunFuncA => stateValue = funcA   //运算funcA，更新stateValue
    case RunFuncB => stateValue = funcB    //运算funcB，更新stateValue
    ...
  }
}

OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
    case _: ArithmeticException      => Resume
    case _: SomeMinerExecption       => Resume 
    case _: NullPointerException     => Restart
    case _: IllegalArgumentException => Stop
    case _: Exception                => Escalate
  }

# The guardian "/user" will use this class to obtain its supervisorStrategy.
# It needs to be a subclass of akka.actor.SupervisorStrategyConfigurator.
# In addition to the default there is akka.actor.StoppingSupervisorStrategy.
    guardian-supervisor-strategy = "akka.actor.DefaultSupervisorStrategy"

/**
   * When supervisorStrategy is not specified for an actor this
   * `Decider` is used by default in the supervisor strategy.
   * The child will be stopped when [[akka.actor.ActorInitializationException]],
   * [[akka.actor.ActorKilledException]], or [[akka.actor.DeathPactException]] is
   * thrown. It will be restarted for other `Exception` types.
   * The error is escalated if it's a `Throwable`, i.e. `Error`.
   */
  final val defaultDecider: Decider = {
    case _: ActorInitializationException ⇒ Stop
    case _: ActorKilledException         ⇒ Stop
    case _: DeathPactException           ⇒ Stop
    case _: Exception                    ⇒ Restart
  }

  /**
   * When supervisorStrategy is not specified for an actor this
   * is used by default. OneForOneStrategy with decider defined in
   * [[#defaultDecider]].
   */
  f
inal val defaultStrategy: SupervisorStrategy = {
    OneForOneStrategy()(defaultDecider)
  }

override val supervisorStrategy =
  OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
    case _: ArithmeticException => Resume
    case _: MyException => Restart  
    case t =>
      super.supervisorStrategy.decider.applyOrElse(t, (_: Any) => Escalate)
  }

/**
   * User overridable callback.
   * <p/>
   * Is called when an Actor is started.
   * Actors are automatically started asynchronously when created.
   * Empty default implementation.
   */
  @throws(classOf[Exception]) // when changing this you MUST also change ActorDocTest
  //#lifecycle-hooks
  def preStart(): Unit = ()

  override def preStart={
    log.info ("Starting storage actor...")
    initDB
  }

/**
   * User overridable callback.
   * <p/>
   * Is called asynchronously after 'actor.stop()' is invoked.
   * Empty default implementation.
   */
  @throws(classOf[Exception]) // when changing this you MUST also change ActorDocTest
  //#lifecycle-hooks
  def postStop(): Unit = ()

  override def postStop={
    log.info ("Stopping storage actor...")
    db.release
  }

  /**
   * Scala API: User overridable callback: '''By default it disposes of all children and then calls `postStop()`.'''
   * @param reason the Throwable that caused the restart to happen
   * @param message optionally the current message the actor processed when failing, if applicable
   * <p/>
   * Is called on a crashed Actor right BEFORE it is restarted to allow clean
   * up of resources before Actor is terminated.
   */
  @throws(classOf[Exception]) // when changing this you MUST also change ActorDocTest
  //#lifecycle-hooks
  def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    context.children foreach { child ⇒
      context.unwatch(child)
      context.stop(child)
    }
    postStop()
  }

  //#lifecycle-hooks

  /**
   * User overridable callback: By default it calls `preStart()`.
   * @param reason the Throwable that caused the restart to happen
   * <p/>
   * Is called right AFTER restart on the newly created Actor to allow reinitialization after an Actor crash.
   */
  @throws(classOf[Exception]) // when changing this you MUST also change ActorDocTest
  //#lifecycle-hooks
  def postRestart(reason: Throwable): Unit = {
    preStart()
  }

  override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    log.info(s"Parent restarting with error ${message}...")
    doSomeWorkBeforeStopping 
    super.preRestart(reason, message)
  }

  override def postRestart(reason: Throwable): Unit = {
    log.info("need to initialize too ...")
    doSomeExtraInit
    super.postRestart(reason)
  }

/**
 * Applies the fault handling `Directive` (Resume, Restart, Stop) specified in the `Decider`
 * to the child actor that failed, as opposed to [[akka.actor.AllForOneStrategy]] that applies
 * it to all children.
 *
 * @param maxNrOfRetries the number of times a child actor is allowed to be restarted, negative value means no limit,
 *   if the limit is exceeded the child actor is stopped
 * @param withinTimeRange duration of the time window for maxNrOfRetries, Duration.Inf means no window
 * @param decider mapping from Throwable to [[akka.actor.SupervisorStrategy.Directive]], you can also use a
 *   [[scala.collection.immutable.Seq]] of Throwables which maps the given Throwables to restarts, otherwise escalates.
 * @param loggingEnabled the strategy logs the failure if this is enabled (true), by default it is enabled
 */
case class OneForOneStrategy(
  maxNrOfRetries:              Int      = -1,
  withinTimeRange:             Duration = Duration.Inf,
  override val loggingEnabled: Boolean  = true)(val decider: SupervisorStrategy.Decider)
  extends SupervisorStrategy {...}

/**
 * Back-off supervisor that stops and starts a child actor using a back-off algorithm when the child actor stops.
 * This back-off supervisor is created by using `akka.pattern.BackoffSupervisor.props`
 * with `Backoff.onStop`.
 */
final class BackoffSupervisor(
  val childProps: Props,
  val childName:  String,
  minBackoff:     FiniteDuration,
  maxBackoff:     FiniteDuration,
  val reset:      BackoffReset,
  randomFactor:   Double,
  strategy:       SupervisorStrategy)
  extends Actor with HandleBackoff {...}
/**
   * Props for creating a [[BackoffSupervisor]] actor from [[BackoffOptions]].
   *
   * @param options the [[BackoffOptions]] that specify how to construct a backoff-supervisor.
   */
  def props(options: BackoffOptions): Props = options.props
/**
 * Builds back-off options for creating a back-off supervisor.
 * You can pass `BackoffOptions` to `akka.pattern.BackoffSupervisor.props`.
 * An example of creating back-off options:
 * {{{
 * Backoff.onFailure(childProps, childName, minBackoff, maxBackoff, randomFactor)
 *               .withManualReset
 *               .withSupervisorStrategy(
 *                 OneforOneStrategy(){
 *                    case e: GivingUpException => Stop
 *                    case e: RetryableException => Restart
 *                 }
 *               )
 *
 * }}}
 */
object Backoff {
  /**
   * Back-off options for creating a back-off supervisor actor that expects a child actor to restart on failure.
   *
   * This explicit supervisor behaves similarly to the normal implicit supervision where
   * if an actor throws an exception, the decider on the supervisor will decide when to
   * `Stop`, `Restart`, `Escalate`, `Resume` the child actor.
   *
   * When the `Restart` directive is specified, the supervisor will delay the restart
   * using an exponential back off strategy (bounded by minBackoff and maxBackoff).
   *
   * This supervisor is intended to be transparent to both the child actor and external actors.
   * Where external actors can send messages to the supervisor as if it was the child and the
   * messages will be forwarded. And when the child is `Terminated`, the supervisor is also
   * `Terminated`.
   * Transparent to the child means that the child does not have to be aware that it is being
   * supervised specifically by this actor. Just like it does
   * not need to know when it is being supervised by the usual implicit supervisors.
   * The only caveat is that the `ActorRef` of the child is not stable, so any user storing the
   * `sender()` `ActorRef` from the child response may eventually not be able to communicate with
   * the stored `ActorRef`. In general all messages to the child should be directed through this actor.
   *
   * An example of where this supervisor might be used is when you may have an actor that is
   * responsible for continuously polling on a server for some resource that sometimes may be down.
   * Instead of hammering the server continuously when the resource is unavailable, the actor will
   * be restarted with an exponentially increasing back off until the resource is available again.
   *
   * '''***
   * This supervisor should not be used with `Akka Persistence` child actors.
   * `Akka Persistence` actors shutdown unconditionally on `persistFailure()`s rather
   * than throw an exception on a failure like normal actors.
   * [[#onStop]] should be used instead for cases where the child actor
   * terminates itself as a failure signal instead of the normal behavior of throwing an exception.
   * ***'''
   * You can define another
   * supervision strategy by using `akka.pattern.BackoffOptions.withSupervisorStrategy` on [[akka.pattern.BackoffOptions]].
   *
   * @param childProps the [[akka.actor.Props]] of the child actor that
   *   will be started and supervised
   * @param childName name of the child actor
   * @param minBackoff minimum (initial) duration until the child actor will
   *   started again, if it is terminated
   * @param maxBackoff the exponential back-off is capped to this duration
   * @param randomFactor after calculation of the exponential back-off an additional
   *   random delay based on this factor is added, e.g. `0.2` adds up to `20%` delay.
   *   In order to skip this additional delay pass in `0`.
   */
  def onFailure(
    childProps:   Props,
    childName:    String,
    minBackoff:   FiniteDuration,
    maxBackoff:   FiniteDuration,
    randomFactor: Double): BackoffOptions =
    BackoffOptionsImpl(RestartImpliesFailure, childProps, childName, minBackoff, maxBackoff, randomFactor)

  /**
   * Back-off options for creating a back-off supervisor actor that expects a child actor to stop on failure.
   *
   * This actor can be used to supervise a child actor and start it again
   * after a back-off duration if the child actor is stopped.
   *
   * This is useful in situations where the re-start of the child actor should be
   * delayed e.g. in order to give an external resource time to recover before the
   * child actor tries contacting it again (after being restarted).
   *
   * Specifically this pattern is useful for persistent actors,
   * which are stopped in case of persistence failures.
   * Just restarting them immediately would probably fail again (since the data
   * store is probably unavailable). It is better to try again after a delay.
   *
   * It supports exponential back-off between the given `minBackoff` and
   * `maxBackoff` durations. For example, if `minBackoff` is 3 seconds and
   * `maxBackoff` 30 seconds the start attempts will be delayed with
   * 3, 6, 12, 24, 30, 30 seconds. The exponential back-off counter is reset
   * if the actor is not terminated within the `minBackoff` duration.
   *
   * In addition to the calculated exponential back-off an additional
   * random delay based the given `randomFactor` is added, e.g. 0.2 adds up to 20%
   * delay. The reason for adding a random delay is to avoid that all failing
   * actors hit the backend resource at the same time.
   *
   * You can retrieve the current child `ActorRef` by sending `BackoffSupervisor.GetCurrentChild`
   * message to this actor and it will reply with [[akka.pattern.BackoffSupervisor.CurrentChild]]
   * containing the `ActorRef` of the current child, if any.
   *
   * The `BackoffSupervisor`delegates all messages from the child to the parent of the
   * `BackoffSupervisor`, with the supervisor as sender.
   *
   * The `BackoffSupervisor` forwards all other messages to the child, if it is currently running.
   *
   * The child can stop itself and send a [[akka.actor.PoisonPill]] to the parent supervisor
   * if it wants to do an intentional stop.
   *
   * Exceptions in the child are handled with the default supervisionStrategy, which can be changed by using
   * [[BackoffOptions#withSupervisorStrategy]] or [[BackoffOptions#withDefaultStoppingStrategy]]. A
   * `Restart` will perform a normal immediate restart of the child. A `Stop` will
   * stop the child, but it will be started again after the back-off duration.
   *
   * @param childProps the [[akka.actor.Props]] of the child actor that
   *   will be started and supervised
   * @param childName name of the child actor
   * @param minBackoff minimum (initial) duration until the child actor will
   *   started again, if it is terminated
   * @param maxBackoff the exponential back-off is capped to this duration
   * @param randomFactor after calculation of the exponential back-off an additional
   *   random delay based on this factor is added, e.g. `0.2` adds up to `20%` delay.
   *   In order to skip this additional delay pass in `0`.
   */
  def onStop(
    childProps:   Props,
    childName:    String,
    minBackoff:   FiniteDuration,
    maxBackoff:   FiniteDuration,
    randomFactor: Double): BackoffOptions =
    BackoffOptionsImpl(StopImpliesFailure, childProps, childName, minBackoff, maxBackoff, randomFactor)
}

/**
 * Configures a back-off supervisor actor. Start with `Backoff.onStop` or `Backoff.onFailure`.
 * BackoffOptions is immutable, so be sure to chain methods like:
 * {{{
 * val options = Backoff.onFailure(childProps, childName, minBackoff, maxBackoff, randomFactor)
 *               .withManualReset
 * context.actorOf(BackoffSupervisor.props(options), name)
 * }}}
 */
trait BackoffOptions {
  /**
   * Returns a new BackoffOptions with automatic back-off reset.
   * The back-off algorithm is reset if the child does not crash within the specified `resetBackoff`.
   * @param resetBackoff The back-off is reset if the child does not crash within this duration.
   */
  def withAutoReset(resetBackoff: FiniteDuration): BackoffOptions

  /**
   * Returns a new BackoffOptions with manual back-off reset. The back-off is only reset
   * if the child sends a `BackoffSupervisor.Reset` to its parent (the backoff-supervisor actor).
   */
  def withManualReset: BackoffOptions

  /**
   * Returns a new BackoffOptions with the supervisorStrategy.
   * @param supervisorStrategy the supervisorStrategy that the back-off supervisor will use.
   *   The default supervisor strategy is used as fallback if the specified supervisorStrategy (its decider)
   *   does not explicitly handle an exception. As the BackoffSupervisor creates a separate actor to handle the
   *   backoff process, only a [[OneForOneStrategy]] makes sense here.
   */
  def withSupervisorStrategy(supervisorStrategy: OneForOneStrategy): BackoffOptions

  /**
   * Returns a new BackoffOptions with a default `SupervisorStrategy.stoppingStrategy`.
   * The default supervisor strategy is used as fallback for throwables not handled by `SupervisorStrategy.stoppingStrategy`.
   */
  def withDefaultStoppingStrategy: BackoffOptions

  /**
   * Returns the props to create the back-off supervisor.
   */
  private[akka] def props: Props
}

val childProps = Props(classOf[EchoActor])

  val supervisor = BackoffSupervisor.props(
    Backoff.onFailure(
      childProps,
      childName = "myEcho",
      minBackoff = 3.seconds,
      maxBackoff = 30.seconds,
      randomFactor = 0.2 // adds 20% "noise" to vary the intervals slightly
    ).withManualReset
      .withSupervisorStrategy(
         OneforOneStrategy(){
           case _: GivingUpException => Stop
           case _: RetryableException => Restart
           case _: MinorException => Resume 
         }
       )
  )
  
  system.actorOf(supervisor, name = "echoSupervisor")

    case Reset ⇒
      reset match {
        case ManualReset ⇒ restartCount = 0
        case msg         ⇒ unhandled(msg)
      }

  /**
   * Returns a new BackoffOptions with automatic back-off reset.
   * The back-off algorithm is reset if the child does not crash within the specified `resetBackoff`.
   * @param resetBackoff The back-off is reset if the child does not crash within this duration.
   */
  def withAutoReset(resetBackoff: FiniteDuration): BackoffOptions

  /**
   * Returns a new BackoffOptions with manual back-off reset. The back-off is only reset
   * if the child sends a `BackoffSupervisor.Reset` to its parent (the backoff-supervisor actor).
   */
  def withManualReset: BackoffOptions

/**
   * Have this FunctionRef watch the given Actor. This method must not be
   * called concurrently from different threads, it should only be called by
   * its parent Actor.
   *
   * Upon receiving the Terminated message, unwatch() must be called from a
   * safe context (i.e. normally from the parent Actor).
   */
  def watch(actorRef: ActorRef): Unit = {
    watching += actorRef
    actorRef.asInstanceOf[InternalActorRef].sendSystemMessage(Watch(actorRef.asInstanceOf[InternalActorRef], this))
  }

  /**
   * Have this FunctionRef unwatch the given Actor. This method must not be
   * called concurrently from different threads, it should only be called by
   * its parent Actor.
   */
  def unwatch(actorRef: ActorRef): Unit = {
    watching -= actorRef
    actorRef.asInstanceOf[InternalActorRef].sendSystemMessage(Unwatch(actorRef.asInstanceOf[InternalActorRef], this))
  }

class DeathPactExceptionParentActor extends Actor with ActorLogging{
  def receive={
    case "create_child"=> {
      log.info ("creating child")
      val child=context.actorOf(Props[DeathPactExceptionChildActor])
      context.watch(child) //watch child's death
      child!"stop"
    }
    case "someMessage" => log.info ("some message")
    case Terminated(_) => context.stop(self)  //child has stopped
  } 
}

object Chef {
  sealed trait Cooking
  case object CookSpecial extends Cooking
  class ChefBusy(msg: String) extends Exception(msg)
  def props = Props(new Chef)
}
class Chef extends Actor with ActorLogging {
  import Chef._
  log.info(s"Chef actor created at ${System.currentTimeMillis()}")
  override def receive: Receive = {
    case _ => throw new ChefBusy("Chef is busy cooking!")
  }

  override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    super.preRestart(reason, message)
    log.info(s"Restarting Chef for $message")
  }

  override def postRestart(reason: Throwable): Unit = {
    super.postRestart(reason)
    log.info(s"Chef restarted for ${reason.getMessage}")
  }

  override def postStop(): Unit = {
    log.info("Chef stopped!")
  }
}

object Kitchen {
  def kitchenProps = {
    import Chef._
    val options = Backoff.onFailure(Chef.props, "chef", 200 millis, 10 seconds, 0.0)
      .withSupervisorStrategy(OneForOneStrategy(maxNrOfRetries = 4, withinTimeRange = 30 seconds) {
        case _: ChefBusy => SupervisorStrategy.Restart
      })
    BackoffSupervisor.props(options)
  }
}
class Kitchen extends Actor with ActorLogging {
  override def receive: Receive = {
    case x => context.children foreach {child => child ! x}
  }
}

object Cafe extends App {
  import Kitchen._
  val cafeSystem = ActorSystem("cafe")

  val kitchen = cafeSystem.actorOf(kitchenProps,"kitchen")

  println(s"Calling chef at ${System.currentTimeMillis()}")
  kitchen ! "CookCook"
  println(s"Calling chef at ${System.currentTimeMillis()}")
  Thread.sleep(1000)
  println(s"Calling chef at ${System.currentTimeMillis()}")
  kitchen ! "CookCook"
  Thread.sleep(1000)
  kitchen ! "CookCook"
  Thread.sleep(1000)
  kitchen ! "CookCook"
  Thread.sleep(1000)
  kitchen ! "CookCook"

  Thread.sleep(1000 * 30)
  cafeSystem.terminate()

}

Calling chef at 1495108529380
[INFO] [05/18/2017 19:55:29.384] [cafe-akka.actor.default-dispatcher-2] [akka://cafe/user/kitchen/chef] Chef actor created at 1495108529382
[ERROR] [05/18/2017 19:55:29.392] [cafe-akka.actor.default-dispatcher-3] [akka://cafe/user/kitchen/chef] Chef is busy cooking!
Chef$ChefBusy: Chef is busy cooking!
    at Chef$$anonfun$receive$1.applyOrElse(Cafe.scala:24)
...
[INFO] [05/18/2017 19:55:29.394] [cafe-akka.actor.default-dispatcher-2] [akka://cafe/user/kitchen/chef] Chef stopped!
[INFO] [05/18/2017 19:55:29.614] [cafe-akka.actor.default-dispatcher-4] [akka://cafe/user/kitchen/chef] Chef actor created at 1495108529614
Calling chef at 1495108530382
[ERROR] [05/18/2017 19:55:30.382] [cafe-akka.actor.default-dispatcher-3] [akka://cafe/user/kitchen/chef] Chef is busy cooking!

import akka.actor._
import scala.util.Random
import scala.concurrent.duration._
object ChildActor {
  class RndException(msg: String) extends Exception(msg)
  def props = Props[ChildActor]
}
class ChildActor extends Actor with ActorLogging {
  import ChildActor._
  override def receive: Receive = {
    case msg: String => {   //任意产生一些RndExcption
      if (Random.nextBoolean()) 
        throw new RndException("Any Exception!")
      else
        log.info(s"Processed message: $msg !!!")
    }
  }

  override def preStart(): Unit = {
    log.info("ChildActor Started.")
    super.preStart()
  }
  //在重启时preRestart是在原来的Actor实例上调用preRestart的
  override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    log.info(s"Restarting ChildActor for ${reason.getMessage}...")
    message match {
      case Some(msg) =>
        log.info(s"Exception message: ${msg.toString}")
        self ! msg       //把异常消息再摆放到信箱最后
      case None =>
    }
    super.preRestart(reason, message)
  }

  override def postRestart(reason: Throwable): Unit = {
    super.postRestart(reason)
    log.info(s"Restarted ChildActor for ${reason.getMessage}...")
  }

  override def postStop(): Unit = {
    log.info(s"Stopped ChildActor.")
    super.postStop()
  }

}

//监管父级
class Parent extends Actor with ActorLogging {

  def decider: PartialFunction[Throwable,SupervisorStrategy.Directive] = {
    case _: ChildActor.RndException => SupervisorStrategy.Restart
  }

  override val supervisorStrategy =
    OneForOneStrategy(maxNrOfRetries = 30, withinTimeRange = 3 seconds) {
      decider.orElse(SupervisorStrategy.defaultDecider)
    }

  val childActor = context.actorOf(ChildActor.props,"childActor")
  override def receive: Receive = {
    case msg@ _ => childActor ! msg    //把所有收到的消息都转给childActor
  }

}

object TestMyActor extends App {
  val system = ActorSystem("testSystem")
  val parentActor = system.actorOf(Props[Parent],"parentActor")

  parentActor ! "Hello 1"
  parentActor ! "Hello 2"
  parentActor ! "Hello 3"
  parentActor ! "Hello 4"
  parentActor ! "Hello 5"

  Thread.sleep(5000)
  system.terminate()
}

object Chef {
  sealed trait Order  //消息类型
  case object MakeSpecial extends Order  //烹制特饮
  class ChefBusy(msg: String) extends Exception(msg)  //异常类型
  def props = Props[Chef]
}
class Chef extends Actor with ActorLogging {
  import Chef._

  log.info("Chef says: I am ready to work ...")   //构建成功信息
  //内部状态
  var currentSpecial: Cafe.Coffee = Cafe.Original
  var chefBusy: Boolean = false

  val specials = Map(0 -> Cafe.Original,1 -> Cafe.Espresso, 2 -> Cafe.Cappuccino)

  override def receive: Receive = {
    case MakeSpecial => {
      if ((Random.nextInt(6) % 6) == 0) {  //任意产生异常 2/6
        log.info("Chef is busy ...")
        chefBusy = true
        throw new ChefBusy("Busy!")
      }
      else {
        currentSpecial = randomSpecial     //选出当前特饮
        log.info(s"Chef says: Current special is ${currentSpecial.toString}.")

        sender() ! currentSpecial
      }
    }
  }
  def randomSpecial = specials(Random.nextInt(specials.size)) //选出当前特饮

  override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    log.info(s"Restarting Chef for ${reason.getMessage}...")
    super.preRestart(reason, message)
  }

  override def postRestart(reason: Throwable): Unit = {
    log.info(s"Restarted Chef for ${reason.getMessage}.")
    context.parent ! BackoffSupervisor.Reset

    super.postRestart(reason)
  }

  override def postStop(): Unit = {
    log.info("Stopped Chef.")
    super.postStop()
  }
}
//Kitchen只是Chef的Backoff监管，没有任何其它功能
class Kitchen extends Actor with ActorLogging {
  override def receive: Receive = {
    //context.children.size == 1，就是chef。 直接把所有消息转发到Chef
    case msg@_ =>  //注意，无法使用Chef ？因为sender不明
      context.children foreach ( chef => chef forward msg)
  }

  override def postStop(): Unit = {
    log.info("Kitchen close!")
    super.postStop()
  }
}
object Kitchen {
  //指定的异常处理策略
  val kitchenDecider: PartialFunction[Throwable, SupervisorStrategy.Directive] = {
    case _: Chef.ChefBusy => SupervisorStrategy.Restart
  }
  def kitchenProps: Props = {  //定义BackoffSupervisor strategy
    val option = Backoff.onFailure(Chef.props,"chef",1 seconds, 5 seconds, 0.0)
      .withManualReset
      .withSupervisorStrategy {
        OneForOneStrategy(maxNrOfRetries = 5, withinTimeRange = 5 seconds) {
          kitchenDecider.orElse(SupervisorStrategy.defaultDecider)
        }
      }
    BackoffSupervisor.props(option)
  }
}

object ReceiptPrinter {
  case class PrintReceipt(sendTo: ActorRef, receipt: Cafe.Receipt)  //print command
  class PaperJamException extends Exception
  def props = Props[ReceiptPrinter]
}
class ReceiptPrinter extends Actor with ActorLogging {
  import ReceiptPrinter._
  var paperJammed: Boolean = false
  override def receive: Receive = {
    case PrintReceipt(customer, receipt) =>    //打印收据并发送给顾客
      if ((Random.nextInt(6) % 6) == 0) {
        log.info("Printer jammed paper ...")
        paperJammed = true
        throw new PaperJamException
      } else {
        log.info(s"Printing receipt $receipt and sending to ${customer.path.name}")
        customer ! receipt
      }
  }

  override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    log.info(s"Restarting ReceiptPrinter for ${reason.getMessage}...")
    super.preRestart(reason, message)
  }

  override def postRestart(reason: Throwable): Unit = {
    log.info(s"Started ReceiptPrinter for ${reason.getMessage}.")
    super.postRestart(reason)
  }

  override def postStop(): Unit = {
    log.info("Stopped ReceiptPrinter.")
    super.postStop()
  }
}
object Cashier {
  case class RingRegister(cup: Cafe.Coffee, customer: ActorRef)  //收款并出具收据

  def props(kitchen: ActorRef) = Props(classOf[Cashier],kitchen)
}
class Cashier(kitchen: ActorRef) extends Actor with ActorLogging {
  import Cashier._
  import ReceiptPrinter._

  context.watch(kitchen)   //监视厨房。如果打烊了就关门歇业
  val printer = context.actorOf(ReceiptPrinter.props,"printer")
  //打印机卡纸后重启策略
  def cashierDecider: PartialFunction[Throwable,SupervisorStrategy.Directive] = {
    case _: PaperJamException => SupervisorStrategy.Restart
  }
  override def supervisorStrategy: SupervisorStrategy =
    OneForOneStrategy(maxNrOfRetries = 5, withinTimeRange = 5 seconds){
      cashierDecider.orElse(SupervisorStrategy.defaultDecider)
    }
  val menu = Map[Cafe.Coffee,Double](Cafe.Original -> 5.50,
    Cafe.Cappuccino -> 12.95, Cafe.Espresso -> 11.80)

  override def receive: Receive = {
    case RingRegister(coffee, customer) => //收款并出具收据
      log.info(s"Producing receipt for a cup of ${coffee.toString}...")
      val amt = menu(coffee)    //计价
      val rcpt = Cafe.Receipt(coffee.toString,amt)
      printer ! PrintReceipt(customer,rcpt)  //打印收据。可能出现卡纸异常
      sender() ! Cafe.Sold(rcpt)  //通知Cafe销售成功  sender === Cafe
    case Terminated(_) =>
      log.info("Cashier says: Oh, kitchen is closed. Let's make the end of day!")
      context.system.terminate()   //厨房打烊，停止营业。
  }
}

object Cafe {
  sealed trait Coffee  //咖啡种类
  case object Original extends Coffee
  case object Espresso extends Coffee
  case object Cappuccino extends Coffee

  case class Receipt(item: String, amt: Double)

  sealed trait Routine
  case object PlaceOrder extends Routine
  case class Sold(receipt: Receipt) extends Routine

}
class Cafe extends Actor with ActorLogging {
  import Cafe._
  import Cashier._

  import context.dispatcher
  implicit val timeout = Timeout(1 seconds)

  var totalAmount: Double = 0.0

  val kitchen = context.actorOf(Kitchen.kitchenProps,"kitchen")
  //Chef可能重启，但path不变。必须直接用chef ? msg，否则经Kitchen转发无法获取正确的sender
  val chef = context.actorSelection("/user/cafe/kitchen/chef")

  val cashier = context.actorOf(Cashier.props(kitchen),"cashier")

  var customer: ActorRef = _     //当前客户

  override def receive: Receive = {

    case Sold(rcpt) =>
      totalAmount += rcpt.amt
      log.info(s"Today's sales is up to $totalAmount")
      customer ! Customer.OrderServed(rcpt)   //send him the order
      if (totalAmount > 100.00) {
        log.info("Asking kichen to clean up ...")
        context.stop(kitchen)
      }
    case PlaceOrder =>
      customer = sender()     //send coffee to this customer
      (for {
        item <- (chef ? Chef.MakeSpecial).mapTo[Coffee]
        sales <- (cashier ? RingRegister(item,sender())).mapTo[Sold]
      } yield(Sold(sales.receipt))).mapTo[Sold]
        .recover {
          case _: AskTimeoutException => Customer.ComebackLater
        }.pipeTo(self)   //send receipt to be added to totalAmount

  }
}
object Customer {
  sealed trait CustomerOrder
  case object OrderSpecial extends CustomerOrder
  case class OrderServed(rcpt: Cafe.Receipt) extends CustomerOrder
  case object ComebackLater extends CustomerOrder
  def props(cafe: ActorRef) = Props(new Customer(cafe))
}
class Customer(cafe: ActorRef) extends Actor with ActorLogging {
  import Customer._
  import context.dispatcher
  override def receive: Receive = {
    case OrderSpecial =>
      log.info("Customer place an order ...")
      cafe ! Cafe.PlaceOrder
    case OrderServed(rcpt) =>
      log.info(s"Customer says: Oh my! got my order ${rcpt.item} for ${rcpt.amt}")
    case ComebackLater =>
      log.info("Customer is not so happy! says: I will be back later!")
      context.system.scheduler.scheduleOnce(1 seconds){cafe ! Cafe.PlaceOrder}
  }
}

object MyCafe extends App {
  import Cafe._
  import Customer._
  import scala.concurrent.ExecutionContext.Implicits.global
  val cafeSys = ActorSystem("cafeSystem")
  val cafe = cafeSys.actorOf(Props[Cafe],"cafe")
  val customer = cafeSys.actorOf(Customer.props(cafe),"customer")

  cafeSys.scheduler.schedule(1 second, 1 second, customer, OrderSpecial)
  
}

package mycafe

import akka.actor._

import scala.concurrent.duration._
import scala.util.Random
import akka.pattern._
import akka.util.Timeout
import scala.concurrent._


object Chef {
  sealed trait Order  //消息类型
  case object MakeSpecial extends Order  //烹制特饮
  class ChefBusy(msg: String) extends Exception(msg)  //异常类型
  def props = Props[Chef]
}
class Chef extends Actor with ActorLogging {
  import Chef._

  log.info("Chef says: I am ready to work ...")   //构建成功信息
  //内部状态
  var currentSpecial: Cafe.Coffee = Cafe.Original
  var chefBusy: Boolean = false

  val specials = Map(0 -> Cafe.Original,1 -> Cafe.Espresso, 2 -> Cafe.Cappuccino)

  override def receive: Receive = {
    case MakeSpecial => {
      if ((Random.nextInt(6) % 6) == 0) {  //任意产生异常 2/6
        log.info("Chef is busy ...")
        chefBusy = true
        throw new ChefBusy("Busy!")
      }
      else {
        currentSpecial = randomSpecial     //选出当前特饮
        log.info(s"Chef says: Current special is ${currentSpecial.toString}.")

        sender() ! currentSpecial
      }
    }
  }
  def randomSpecial = specials(Random.nextInt(specials.size)) //选出当前特饮

  override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    log.info(s"Restarting Chef for ${reason.getMessage}...")
    super.preRestart(reason, message)
  }

  override def postRestart(reason: Throwable): Unit = {
    log.info(s"Restarted Chef for ${reason.getMessage}.")
    context.parent ! BackoffSupervisor.Reset

    super.postRestart(reason)
  }

  override def postStop(): Unit = {
    log.info("Stopped Chef.")
    super.postStop()
  }
}
//Kitchen只是Chef的Backoff监管，没有任何其它功能
class Kitchen extends Actor with ActorLogging {
  override def receive: Receive = {
    //context.children.size == 1，就是chef。 直接把所有消息转发到Chef
    case msg@_ =>  //注意，无法使用Chef ？因为sender不明
      context.children foreach ( chef => chef forward msg)
  }

  override def postStop(): Unit = {
    log.info("Kitchen close!")
    super.postStop()
  }
}
object Kitchen {
  //指定的异常处理策略
  val kitchenDecider: PartialFunction[Throwable, SupervisorStrategy.Directive] = {
    case _: Chef.ChefBusy => SupervisorStrategy.Restart
  }
  def kitchenProps: Props = {  //定义BackoffSupervisor strategy
    val option = Backoff.onFailure(Chef.props,"chef",1 seconds, 5 seconds, 0.0)
      .withManualReset
      .withSupervisorStrategy {
        OneForOneStrategy(maxNrOfRetries = 5, withinTimeRange = 5 seconds) {
          kitchenDecider.orElse(SupervisorStrategy.defaultDecider)
        }
      }
    BackoffSupervisor.props(option)
  }
}


object ReceiptPrinter {
  case class PrintReceipt(sendTo: ActorRef, receipt: Cafe.Receipt)  //print command
  class PaperJamException extends Exception
  def props = Props[ReceiptPrinter]
}
class ReceiptPrinter extends Actor with ActorLogging {
  import ReceiptPrinter._
  var paperJammed: Boolean = false
  override def receive: Receive = {
    case PrintReceipt(customer, receipt) =>    //打印收据并发送给顾客
      if ((Random.nextInt(6) % 6) == 0) {
        log.info("Printer jammed paper ...")
        paperJammed = true
        throw new PaperJamException
      } else {
        log.info(s"Printing receipt $receipt and sending to ${customer.path.name}")
        customer ! receipt
      }
  }

  override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
    log.info(s"Restarting ReceiptPrinter for ${reason.getMessage}...")
    super.preRestart(reason, message)
  }

  override def postRestart(reason: Throwable): Unit = {
    log.info(s"Started ReceiptPrinter for ${reason.getMessage}.")
    super.postRestart(reason)
  }

  override def postStop(): Unit = {
    log.info("Stopped ReceiptPrinter.")
    super.postStop()
  }
}
object Cashier {
  case class RingRegister(cup: Cafe.Coffee, customer: ActorRef)  //收款并出具收据

  def props(kitchen: ActorRef) = Props(classOf[Cashier],kitchen)
}
class Cashier(kitchen: ActorRef) extends Actor with ActorLogging {
  import Cashier._
  import ReceiptPrinter._

  context.watch(kitchen)   //监视厨房。如果打烊了就关门歇业
  val printer = context.actorOf(ReceiptPrinter.props,"printer")
  //打印机卡纸后重启策略
  def cashierDecider: PartialFunction[Throwable,SupervisorStrategy.Directive] = {
    case _: PaperJamException => SupervisorStrategy.Restart
  }
  override def supervisorStrategy: SupervisorStrategy =
    OneForOneStrategy(maxNrOfRetries = 5, withinTimeRange = 5 seconds){
      cashierDecider.orElse(SupervisorStrategy.defaultDecider)
    }
  val menu = Map[Cafe.Coffee,Double](Cafe.Original -> 5.50,
    Cafe.Cappuccino -> 12.95, Cafe.Espresso -> 11.80)

  override def receive: Receive = {
    case RingRegister(coffee, customer) => //收款并出具收据
      log.info(s"Producing receipt for a cup of ${coffee.toString}...")
      val amt = menu(coffee)    //计价
      val rcpt = Cafe.Receipt(coffee.toString,amt)
      printer ! PrintReceipt(customer,rcpt)  //打印收据。可能出现卡纸异常
      sender() ! Cafe.Sold(rcpt)  //通知Cafe销售成功  sender === Cafe
    case Terminated(_) =>
      log.info("Cashier says: Oh, kitchen is closed. Let's make the end of day!")
      context.system.terminate()   //厨房打烊，停止营业。
  }
}

object Cafe {
  sealed trait Coffee  //咖啡种类
  case object Original extends Coffee
  case object Espresso extends Coffee
  case object Cappuccino extends Coffee

  case class Receipt(item: String, amt: Double)

  sealed trait Routine
  case object PlaceOrder extends Routine
  case class Sold(receipt: Receipt) extends Routine

}
class Cafe extends Actor with ActorLogging {
  import Cafe._
  import Cashier._

  import context.dispatcher
  implicit val timeout = Timeout(1 seconds)

  var totalAmount: Double = 0.0

  val kitchen = context.actorOf(Kitchen.kitchenProps,"kitchen")
  //Chef可能重启，但path不变。必须直接用chef ? msg，否则经Kitchen转发无法获取正确的sender
  val chef = context.actorSelection("/user/cafe/kitchen/chef")

  val cashier = context.actorOf(Cashier.props(kitchen),"cashier")

  var customer: ActorRef = _     //当前客户

  override def receive: Receive = {

    case Sold(rcpt) =>
      totalAmount += rcpt.amt
      log.info(s"Today's sales is up to $totalAmount")
      customer ! Customer.OrderServed(rcpt)   //send him the order
      if (totalAmount > 100.00) {
        log.info("Asking kichen to clean up ...")
        context.stop(kitchen)
      }
    case PlaceOrder =>
      customer = sender()     //send coffee to this customer
      (for {
        item <- (chef ? Chef.MakeSpecial).mapTo[Coffee]
        sales <- (cashier ? RingRegister(item,sender())).mapTo[Sold]
      } yield(Sold(sales.receipt))).mapTo[Sold]
        .recover {
          case _: AskTimeoutException => Customer.ComebackLater
        }.pipeTo(self)   //send receipt to be added to totalAmount

  }
}
object Customer {
  sealed trait CustomerOrder
  case object OrderSpecial extends CustomerOrder
  case class OrderServed(rcpt: Cafe.Receipt) extends CustomerOrder
  case object ComebackLater extends CustomerOrder
  def props(cafe: ActorRef) = Props(new Customer(cafe))
}
class Customer(cafe: ActorRef) extends Actor with ActorLogging {
  import Customer._
  import context.dispatcher
  override def receive: Receive = {
    case OrderSpecial =>
      log.info("Customer place an order ...")
      cafe ! Cafe.PlaceOrder
    case OrderServed(rcpt) =>
      log.info(s"Customer says: Oh my! got my order ${rcpt.item} for ${rcpt.amt}")
    case ComebackLater =>
      log.info("Customer is not so happy! says: I will be back later!")
      context.system.scheduler.scheduleOnce(1 seconds){cafe ! Cafe.PlaceOrder}
  }
}


object MyCafe extends App {
  import Cafe._
  import Customer._
  import scala.concurrent.ExecutionContext.Implicits.global
  val cafeSys = ActorSystem("cafeSystem")
  val cafe = cafeSys.actorOf(Props[Cafe],"cafe")
  val customer = cafeSys.actorOf(Customer.props(cafe),"customer")

  cafeSys.scheduler.schedule(1 second, 1 second, customer, OrderSpecial)

}