为长期存活的Akka Actor实现轻微异常重试及监管策略

我之前在项目里处理过几乎一模一样的Akka Actor重试场景，结合你的需求，给你一套落地性很强的方案：

把重试逻辑放在子Actor内部（针对可重试的MinorException），只有当重试次数耗尽后，才触发父Actor的监管重启；对于不可重试的异常，直接抛出让父Actor立即重启。这样既避免了不必要的Actor重启，又能保证异常场景下的容错性。

1. 定义消息和异常类型 #

首先明确你的异常分类，并用包装消息携带重试次数：

// 自定义轻微异常，标记可重试的场景
case class MinorException(message: String) extends Exception(message)

// 包装原始触发消息，携带当前重试次数（默认从0开始）
case class RetryableFileFetch(attempt: Int = 0)
// 保留原始触发消息，兼容父Actor的调用逻辑
case object FileFetch

2. 重构子Actor的消息处理逻辑 #

子Actor内部处理重试判断、延迟重试，仅在重试耗尽或遇到不可重试异常时抛出错误：

import akka.actor.{Actor, ActorLogging, Props}
import scala.concurrent.duration._
import scala.util.{Failure, Success}

class SftpFileFetcherActor extends Actor with ActorLogging {
  import context.dispatcher
  
  // 可配置的重试参数
  private val MaxRetries = 3       // 最大重试次数
  private val RetryDelay = 5.minutes // 重试间隔（避免频繁冲击SFTP服务器）

  override def receive: Receive = {
    // 接收父Actor的原始触发消息，转成带重试计数的消息开始处理
    case FileFetch => self ! RetryableFileFetch(0)
    
    case RetryableFileFetch(attempt) =>
      log.info(s"SFTP文件拉取尝试：第${attempt + 1}/${MaxRetries}次")
      
      // 执行SFTP拉取操作（这里假设是异步Future，同步操作可改用try-catch）
      val fetchTask = performSftpFileFetch()
      
      fetchTask.onComplete {
        case Success(_) =>
          log.info("SFTP文件拉取成功")
        
        case Failure(e: MinorException) =>
          if (attempt < MaxRetries - 1) {
            log.warning(s"轻微异常触发重试：${e.getMessage}，${RetryDelay}后重试")
            // 延迟发送下一次重试请求
            context.system.scheduler.scheduleOnce(RetryDelay, self, RetryableFileFetch(attempt + 1))
          } else {
            log.error(s"${MaxRetries}次重试后仍失败，触发Actor重启")
            // 重试耗尽，抛出异常交给父Actor监管处理
            throw e
          }
        
        case Failure(e) =>
          log.error(s"不可重试异常：${e.getMessage}，立即触发Actor重启")
          // 非轻微异常直接抛出，让父Actor立即重启
          throw e
      }
  }

  // 替换成你的实际SFTP拉取逻辑
  private def performSftpFileFetch(): Future[Unit] = {
    // 这里写你的SFTP连接、文件拉取代码
    ???
  }
}

3. 调整父Actor的监管策略（可选但更清晰） #

你的原有监管策略可以保留，但如果想更精准区分异常类型，可修改为：

import akka.actor.{OneForOneStrategy, SupervisorStrategy, Actor}
import scala.concurrent.duration.Duration

class ParentActor extends Actor {
  private val sftpFetcher = context.actorOf(Props[SftpFileFetcherActor], "sftp-fetcher")

  override val supervisorStrategy: SupervisorStrategy = OneForOneStrategy(
    maxNrOfRetries = -1, 
    withinTimeRange = Duration.Inf
  ) {
    case _: MinorException => SupervisorStrategy.Restart // 重试耗尽后的轻微异常
    case _: Exception => SupervisorStrategy.Restart     // 所有其他异常
  }

  // 每日触发逻辑（省略定时调度代码，比如用Akka Scheduler实现每日触发）
  override def receive: Receive = {
    case DailyTrigger => sftpFetcher ! FileFetch
  }
}

• 重试逻辑放在子Actor内部，避免了频繁重启Actor导致的状态丢失；
• 重试间隔设置合理延迟，防止短时间内重复请求SFTP服务器加重负载；
• 仅当重试耗尽后才触发重启，符合你对轻微异常的容错需求；
• 不可重试异常直接抛出，保证错误场景下快速重启恢复。

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