从Actor获取响应时遇死信，Mongo异常致ES结果无法返回

首先咱们得搞清楚为什么会出现死信、拿不到ES结果的问题：

当MongoActor抛出MongoException时，你的ParentActor在mFuture.onComplete的Failure分支里执行了throw e——这可是个致命操作！Akka Actor的规则是：如果在Actor的处理逻辑里抛出未捕获的异常，Actor系统会触发重启机制，当前的ParentActor实例会被终止并重启。这直接导致两个问题：

1. 正在处理的ES请求的回复上下文丢失了——原来的ParentActor已经不存在，esActor处理完返回的结果找不到收件人，只能变成死信。
2. Demo里的esFuture会一直处于等待状态（你用了Duration.Inf），永远拿不到ES的处理结果。

修复方案 #

咱们需要从两个核心点入手：阻止ParentActor崩溃，确保ES请求的回复能正常传递到Demo。

1. 修改ParentActor：移除致命的throw e，规范错误处理

import akka.actor._
import scala.concurrent.Future
import scala.util.{Success, Failure}

class ParentActor extends Actor with ActorLogging {
  val mongoActor = context.actorOf(Props[MongoActor], "mongoActor")
  val esActor = context.actorOf(Props[EsActor], "esActor")

  // 可选：配置子Actor的监督策略，优雅处理子Actor异常
  override def supervisorStrategy: SupervisorStrategy = OneForOneStrategy() {
    case _: MongoException => SupervisorStrategy.Restart // 按需选择重启/继续/停止
    case _: Exception => SupervisorStrategy.Restart
  }

  def receive: Receive = {
    case InsertInMongo(obj) =>
      // 提前保存sender引用，避免Future回调中上下文失效
      val originalSender = sender()
      val mFuture = ask(mongoActor, InsertDataInMongo(obj)).mapTo[Boolean]
      mFuture.onComplete {
        case Success(resultMongo) => originalSender ! resultMongo
        case Failure(e) =>
          originalSender ! Status.Failure(e)
          // 只记录日志，不要抛出异常！
          log.error(e, "Failed to insert data into MongoDB")
      }(context.dispatcher) // 显式指定Dispatcher，符合Akka最佳实践
    case InsertInES(obj) =>
      val originalSender = sender()
      val eFuture = ask(esActor, InsertDataInES(obj)).mapTo[Boolean]
      eFuture.onComplete {
        case Success(resultES) => originalSender ! resultES
        case Failure(e) =>
          originalSender ! Status.Failure(e)
          log.error(e, "Failed to insert data into Elasticsearch")
      }(context.dispatcher)
  }
}

2. 优化Demo代码（可选，增强健壮性）

import akka.actor._
import scala.concurrent.{Await, Future}
import scala.concurrent.duration.Duration
import scala.util.control.NonFatal

class Demo(actorSystem: ActorSystem) extends Logging {
  val parentActor = actorSystem.actorOf(Props[ParentActor], "parentActor")
  
  def run(obj: Any): Unit = {
    import actorSystem.dispatcher
    
    val mongoFuture = ask(parentActor, InsertInMongo(obj))
      .mapTo[Boolean]
      .recover {
        case NonFatal(e) => 
          log.error(e, "Mongo insertion failed")
          false
      }
    
    val esFuture = ask(parentActor, InsertInES(obj))
      .mapTo[Boolean]
      .recover {
        case NonFatal(e) => 
          log.error(e, "ES insertion failed")
          false
      }
    
    val combinedResult = Await.result(Future.sequence(List(mongoFuture, esFuture)), Duration.Inf)
    log.info("Final result: {}", combinedResult)
  }
}

关键修改点说明 #

• 移除throw e：这是解决问题的核心！不再让ParentActor崩溃，它就能正常接收esActor的回复，把结果转发给Demo里的请求方。
• 保存sender引用：在Future回调中直接用sender()是不安全的——回调可能在Actor上下文之外执行，sender()会失效。提前保存originalSender能确保回复准确送达。
• 添加监督策略：通过Akka的监督机制处理子Actor的异常（比如MongoActor崩溃），而不是让异常蔓延到ParentActor。你可以根据业务需求调整策略（比如重启、继续运行或停止子Actor）。
• 显式指定Dispatcher：确保Future回调在Actor的Dispatcher上执行，避免线程上下文混乱。

最终效果 #

现在即使Mongo插入失败抛出异常：

1. ParentActor会给Demo返回Status.Failure，Demo的recover分支捕获异常并返回false。
2. ParentActor不会崩溃，能正常接收esActor的回复，把ES的结果（成功或失败）传递给Demo。
3. 再也不会出现死信日志，你能同时拿到Mongo和ES的处理结果。

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