监管与容错
探索 Akka 的监管策略，创建能够优雅地从故障中恢复的弹性系统。
监管与容错 是 CoddyKit 上的免费 Scala for Backend Engineering & Functional Programming 课时。 这是第 3 节课，共 3 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 Scala for Backend Engineering & Functional Programming 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Scala for Backend Engineering & Functional Programming 课程共包含 3 节课。
构建弹性系统
在并发应用程序中，事情可能会出错。Actor 可能崩溃，网络调用可能失败，或者数据库可能无法访问。
容错能力是指即使系统的某些部分发生故障，系统仍能继续正确运行的能力。Akka 提供了实现这一目标的强大机制。
父子监管
Akka Actor 按层次结构组织，非常像家族树。当一个 Actor 创建另一个 Actor 时，它就成为后者的父级。
- 父 Actor 负责监管其子 Actor。
- 如果子 Actor 发生故障，父 Actor 会收到通知。
- 随后，父 Actor 决定如何处理子 Actor 的故障。这称为监管策略。
决定如何处理故障
当 Actor 发生故障时，其父级可以发出以下四种指令之一：
Resume：Actor 忽略该故障并继续处理消息。Restart：Actor 停止、重新创建，然后继续处理消息。其内部状态会被重置。Stop：Actor 被永久终止。Escalate：故障被向上交给父级的监管者（即祖父级）。
OneForOne 策略
OneForOneStrategy 是一种常见的监管策略。它表示指令只应用于发生故障的那个子 Actor。
例如，如果一个子 Actor 崩溃，父 Actor 可以决定只重启该子 Actor，而不影响其他子 Actor。
OneForOne 实战
让我们看看 OneForOneStrategy。我们将创建一个父 Actor：当子 Actor 发生 ArithmeticException 时重启它，而发生其他异常时停止它。
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
您运行上一段代码了吗？下面是运行结果：
- 发送
"divideByZero"时，子 Actor 抛出了ArithmeticException。父 Actor 执行了Restart。您看到了“重启前”和“重启后”的日志。 - 随后，子 Actor 处理了“再次问候”。
- 发送
"nullPointer"时，子 Actor 抛出了NullPointerException。父 Actor 执行了Stop。子 Actor 被终止。 - 之后发送了“空指针之后的问候”，但子 Actor 已经停止，无法处理该消息。
AllForOne 策略
有时，一个子 Actor 的故障意味着所有同级子 Actor 也可能受到影响，或无法正常运行。
AllForOneStrategy 会将相同的指令应用于监管者的所有子 Actor，而不仅仅是发生故障的那个。如果一个子 Actor 发生故障，所有子 Actor 都会受到所选指令的影响。
AllForOne 实战
让我们修改父 Actor，使其使用 AllForOneStrategy。请注意，一个子 Actor 的故障会如何影响其同级 Actor。
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()
}自定义决策器
您可以使用决策函数创建高度定制化的监管逻辑。该函数接收一个 Throwable（异常），并返回一个 Directive。
这样，您就可以根据异常类型、导致异常的消息，甚至监管者的状态来实现复杂规则。
高级监管
下面介绍如何在监管策略中定义自定义决策器，以实现细粒度控制。
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()
}检查您的知识
请思考一个监管两个子 Actor（childA 和 childB）的父 Actor。该父 Actor 使用 AllForOneStrategy，并针对所有异常采用 Restart 指令。
如果 childA 抛出异常，会发生什么？
回顾：容错能力
做得很好！在本课中，您学习了 Akka 强大的监管机制：
- 监管层次结构：父 Actor 监管子 Actor。
- 指令：
Resume、Restart、Stop、Escalate。 OneForOneStrategy：仅将指令应用于发生故障的子 Actor。AllForOneStrategy：将指令应用于所有子 Actor。- 自定义决策器：用于对故障处理进行细粒度控制。
这些工具对于使用 Akka Actor 构建可靠且具有弹性的并发应用程序至关重要！
常见问题解答
「监管与容错」课时是免费的吗？
是的 — 「监管与容错」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Scala for Backend Engineering & Functional Programming 课程的其余内容，请升级到 CoddyKit PRO。 Scala for Backend Engineering & Functional Programming 课程共包含 3 节课。
「监管与容错」这节课中我会学到什么？
探索 Akka 的监管策略，创建能够优雅地从故障中恢复的弹性系统。 你通过在浏览器中直接运行的动手代码来练习 Scala for Backend Engineering & Functional Programming，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 Scala for Backend Engineering & Functional Programming 需要有经验吗？
无需任何先前经验。CoddyKit 上的 Scala for Backend Engineering & Functional Programming 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 3 节课，共 3 节。
「监管与容错」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 Scala for Backend Engineering & Functional Programming 课中编写并运行代码吗？
能。每节 Scala for Backend Engineering & Functional Programming 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。