基于BackoffSupervisor的子Actor失败消息重传重启策略问题

问题分析 #

首先，你的preRestart钩子没有触发，核心原因是BackoffSupervisor的onFailure策略在处理子Actor失败时，会先完全停止旧的子Actor实例（触发postStop），然后等待退避时间再创建一个全新的子Actor（触发preStart），而非执行Akka标准的Restart流程——标准Restart会触发旧实例的preRestart，但退避重启的流程跳过了这一步，所以你看不到对应的日志。

其次，你试图在子Actor内部重发消息的思路不可靠：旧实例停止后就会被销毁，即使触发了preRestart，发送给自己的消息也无法被即将启动的新实例接收。正确的做法是让Supervisor负责捕获、保存失败消息，并在子Actor重启成功后重新发送。

修正方案 #

我们通过自定义Supervisor的逻辑，结合BackoffSupervisor的特性，实现消息的可靠重发和重试次数控制：

1. 定义辅助类与自定义Supervisor

// 封装待重发的消息及重试次数
case class RetryMessage(cmd: NewMail, retryCount: Int = 0)

// 自定义BackoffSupervisor，添加消息重发逻辑
class CustomBackoffSupervisor(childProps: Props, backoffOpts: BackoffOptions) extends BackoffSupervisor(childProps, backoffOpts) {
  // 保存待重发的消息队列
  private var pendingMessages: List[RetryMessage] = Nil

  override def preStart(): Unit = {
    super.preStart()
    // 监听子Actor的生命周期，确保重启后能及时发送消息
    child.foreach(context.watch)
  }

  override def receive: Receive = {
    // 接收从SupervisorStrategy传递来的待重发消息
    case msg@RetryMessage(cmd, count) if count < 3 =>
      pendingMessages = msg :: pendingMessages
      println(s"保存待重发消息，当前重试次数: $count")
    
    // 子Actor终止事件，交给父类处理退避重启逻辑
    case Terminated(childRef) =>
      super.receive(Terminated(childRef))
    
    // 子Actor已重启成功，先处理新消息，再发送待重发消息
    case incomingMsg if child.isDefined =>
      child.get ! incomingMsg
      pendingMessages.foreach { retryMsg =>
        println(s"重发消息，重试次数: ${retryMsg.retryCount}")
        child.get ! retryMsg.cmd
      }
      pendingMessages = Nil
    
    // 其他消息交给父类处理
    case other => super.receive(other)
  }
}

2. 修改Supervisor的初始化代码

val childProps = Props(new SenderActor())
val backoffOpts = Backoff.onFailure(
  childProps,
  childName = cmd.hashCode.toString,
  minBackoff = 1.seconds,
  maxBackoff = 2.seconds,
  randomFactor = 0.2
).withSupervisorStrategy(
  OneForOneStrategy(maxNrOfRetries = 3, loggingEnabled = true) {
    case msgEx: MessageException =>
      println("捕获到消息处理异常")
      // 从自定义异常中提取原始消息，发送给Supervisor保存
      context.self ! RetryMessage(msgEx.cmd)
      SupervisorStrategy.Restart
    case _: Exception =>
      SupervisorStrategy.Restart
    case _ => SupervisorStrategy.Escalate
  }
)

// 创建自定义Supervisor实例
val sup = context.actorOf(Props(new CustomBackoffSupervisor(childProps, backoffOpts)))
sup ! cmd

3. 简化子Actor代码

不需要依赖preRestart，只需要正常处理消息并抛出异常即可：

class SenderActor() extends Actor {
  def fakeSendMail(): Unit = {
    Thread.sleep(1000)
    throw new Exception("模拟发送失败")
  }

  override def preStart(): Unit = {
    println("子Actor启动")
  }

  override def postStop(): Unit = {
    println("子Actor停止")
  }

  override def receive: Receive = {
    case cmd: NewMail =>
      println(s"收到消息: $cmd")
      try {
        fakeSendMail()
      } catch {
        case t => throw MessageException(cmd, t)
      }
  }
}

关键逻辑说明 #

1. 消息捕获与保存：当子Actor抛出MessageException时，SupervisorStrategy从异常中提取原始消息，封装为RetryMessage发送给自定义Supervisor保存。
2. 重试次数控制：通过RetryMessage的retryCount字段，限制最多重试3次，超过次数后不再保存消息。
3. 重启后重发：自定义Supervisor监听子Actor的生命周期，当子Actor重启成功后，自动将待重发消息发送给新的子Actor实例。

内容的提问来源于stack exchange，提问作者TheMP