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) 
}