Akka 主管策略 - 正确的用例
Akka Supervisor Strategy - Correct Use Case
一直在使用Akka Supervisor Strategy来处理业务逻辑异常
阅读最著名的 Scala 博客系列之一 Neophyte,我发现他对我一直在做的事情给出了不同的目的。
示例:
假设我有一个应该联系外部资源的 HttpActor,万一它挂了,我会抛出一个异常,现在 ResourceUnavailableException
.
万一我的主管发现了,我会在我的 HttpActor 上调用 Restart,在我的 HttpActor preRestart
方法中,我会调用 do a schedulerOnce
来重试。
演员:
class HttpActor extends Actor with ActorLogging {
implicit val system = context.system
override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
log.info(s"Restarting Actor due: ${reason.getCause}")
message foreach { msg =>
context.system.scheduler.scheduleOnce(10.seconds, self, msg)
}
}
def receive = LoggingReceive {
case g: GetRequest =>
doRequest(http.doGet(g), g.httpManager.url, sender())
}
一位主管:
class HttpSupervisor extends Actor with ActorLogging with RouterHelper {
override val supervisorStrategy =
OneForOneStrategy(maxNrOfRetries = 5) {
case _: ResourceUnavailableException => Restart
case _: Exception => Escalate
}
var router = makeRouter[HttpActor](5)
def receive = LoggingReceive {
case g: GetRequest =>
router.route(g, sender())
case Terminated(a) =>
router = router.removeRoutee(a)
val r = context.actorOf(Props[HttpActor])
context watch r
router = router.addRoutee(r)
}
}
这里有什么意义?
万一我的 doRequest
方法抛出 ResourceUnavailableException
,主管会得到它并重新启动 actor,根据调度程序强制它在一段时间后重新发送消息。我看到的优势是我可以免费获得重试次数和处理异常本身的好方法。
现在看看博客,他展示了一种不同的方法,以防您需要重试,只需发送这样的消息:
def receive = {
case EspressoRequest =>
val receipt = register ? Transaction(Espresso)
receipt.map((EspressoCup(Filled), _)).recover {
case _: AskTimeoutException => ComebackLater
} pipeTo(sender)
case ClosingTime => context.system.shutdown()
}
在 Future
的 AskTimeoutException
的情况下,他将结果作为 ComebackLater
对象通过管道传输,他将处理该对象:
case ComebackLater =>
log.info("grumble, grumble")
context.system.scheduler.scheduleOnce(300.millis) {
coffeeSource ! EspressoRequest
}
对我来说,这几乎就是您可以使用策略主管执行的操作,但是以手动方式进行,没有内置的重试次数逻辑。
那么最好的方法是什么?为什么?我使用akka supervisor策略的概念是完全错误的吗?
您可以使用 BackoffSupervisor
:
Provided as a built-in pattern the akka.pattern.BackoffSupervisor
implements the so-called exponential backoff supervision strategy, starting a child actor again when it fails, each time with a growing time delay between restarts.
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
).withAutoReset(10.seconds) // the child must send BackoffSupervisor.Reset to its parent
.withSupervisorStrategy(
OneForOneStrategy() {
case _: MyException => SupervisorStrategy.Restart
case _ => SupervisorStrategy.Escalate
}))
一直在使用Akka Supervisor Strategy来处理业务逻辑异常
阅读最著名的 Scala 博客系列之一 Neophyte,我发现他对我一直在做的事情给出了不同的目的。
示例:
假设我有一个应该联系外部资源的 HttpActor,万一它挂了,我会抛出一个异常,现在 ResourceUnavailableException
.
万一我的主管发现了,我会在我的 HttpActor 上调用 Restart,在我的 HttpActor preRestart
方法中,我会调用 do a schedulerOnce
来重试。
演员:
class HttpActor extends Actor with ActorLogging {
implicit val system = context.system
override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
log.info(s"Restarting Actor due: ${reason.getCause}")
message foreach { msg =>
context.system.scheduler.scheduleOnce(10.seconds, self, msg)
}
}
def receive = LoggingReceive {
case g: GetRequest =>
doRequest(http.doGet(g), g.httpManager.url, sender())
}
一位主管:
class HttpSupervisor extends Actor with ActorLogging with RouterHelper {
override val supervisorStrategy =
OneForOneStrategy(maxNrOfRetries = 5) {
case _: ResourceUnavailableException => Restart
case _: Exception => Escalate
}
var router = makeRouter[HttpActor](5)
def receive = LoggingReceive {
case g: GetRequest =>
router.route(g, sender())
case Terminated(a) =>
router = router.removeRoutee(a)
val r = context.actorOf(Props[HttpActor])
context watch r
router = router.addRoutee(r)
}
}
这里有什么意义?
万一我的 doRequest
方法抛出 ResourceUnavailableException
,主管会得到它并重新启动 actor,根据调度程序强制它在一段时间后重新发送消息。我看到的优势是我可以免费获得重试次数和处理异常本身的好方法。
现在看看博客,他展示了一种不同的方法,以防您需要重试,只需发送这样的消息:
def receive = {
case EspressoRequest =>
val receipt = register ? Transaction(Espresso)
receipt.map((EspressoCup(Filled), _)).recover {
case _: AskTimeoutException => ComebackLater
} pipeTo(sender)
case ClosingTime => context.system.shutdown()
}
在 Future
的 AskTimeoutException
的情况下,他将结果作为 ComebackLater
对象通过管道传输,他将处理该对象:
case ComebackLater =>
log.info("grumble, grumble")
context.system.scheduler.scheduleOnce(300.millis) {
coffeeSource ! EspressoRequest
}
对我来说,这几乎就是您可以使用策略主管执行的操作,但是以手动方式进行,没有内置的重试次数逻辑。
那么最好的方法是什么?为什么?我使用akka supervisor策略的概念是完全错误的吗?
您可以使用 BackoffSupervisor
:
Provided as a built-in pattern the
akka.pattern.BackoffSupervisor
implements the so-called exponential backoff supervision strategy, starting a child actor again when it fails, each time with a growing time delay between restarts.
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
).withAutoReset(10.seconds) // the child must send BackoffSupervisor.Reset to its parent
.withSupervisorStrategy(
OneForOneStrategy() {
case _: MyException => SupervisorStrategy.Restart
case _ => SupervisorStrategy.Escalate
}))