在两个演员之间使用 pipeTo 时收到 akka 死信

问题描述

parent -> childABC -> workerchild

现在工人孩子工作并将其结果发送给其父母（childABC，它是父母的孩子），而该子演员（childABC）将结果发送回父演员我正在使用pipeto并收到死信这是我的代码

parent 演员：

final case object GetFinalValue

class MyActor extends Actor{
  import context.dispatcher
  import akka.pattern.pipe
  val log = LoggerFactory.getLogger(this.getClass)
  val myManageActor = context.actorOf(Props[ManagerMyActor],"Managemyactor")
  implicit val timeout = Timeout(ReadTimeIntervalValue.getInterval(),SECONDS)

  override def receive: Receive = {
    case GetFinalValue=>
      ask(myManageActor,GetValue).pipeto(sender())

    case message =>
      log.warn(" Unhandled message received : {}",message)
      unhandled(message)
  }

}

childABC（根据我上面给出的例子）

final case object GetValue

class ManagerMyActor extends Actor{
  import context.dispatcher
  import akka.pattern.pipe
  val log = LoggerFactory.getLogger(this.getClass)
  val myTokenActor = context.actorOf(Props[TokenMyActor2],"toknMyActor2")
  implicit val timeout = Timeout(ReadTimeIntervalValue.getInterval(),SECONDS)

  override def receive: Receive = {
    case GetValue=>
      ask(myTokenActor,CalculateValue).pipeto(sender())
 
    case message =>
      log.warn(" Unhandled message received : {}",message)
      unhandled(message)
  }

}

child 演员：

final case object CalculateValue

class TokenMyActor2 extends Actor{
  import context.dispatcher
  import akka.pattern.pipe
  val log = LoggerFactory.getLogger(this.getClass)

  override def receive: Receive = {
    case CalculateValue=>
      val future = Future{ "get the string"
      }
      val bac = future.map{result =>
          sender ! result
      }//.pipeto(sender())


    case message =>
      log.warn("Actor MyActor: Unhandled message received : {}",message)
      unhandled(message)
  }

}


def main(args: Array[String]): Unit = {
    implicit val timeout = Timeout(ReadTimeIntervalValue.getInterval(),SECONDS)

    val myActor = system.actorOf(Props[MyActor],"myActor")
    val future = ask(myActor,GetFinalValue).mapTo[String]
    future.map {str =>
      log.info ("string is {}",str)
    }

这里是日志：

[INFO] [akkaDeadLetter][01/12/2021 19:17:22.000] [api-akka.actor.default-dispatcher-5] [akka://api/deadLetters] Message [@R_502[email protected]] from Actor[akka://api/user/myActor/Managemyactor/toknMyActor2#1239397461] to Actor[akka://api/deadLetters] was not delivered. [1] dead letters encountered. If this is not an expected behavior then Actor[akka://api/deadLetters] may have terminated unexpectedly. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [akkaDeadLetter][01/12/2021 19:17:41.989] [api-akka.actor.default-dispatcher-7] [akka://api/deadLetters] Message [akka.actor.Status$Failure] from Actor[akka://api/user/myActor#1829301550] to Actor[akka://api/deadLetters] was not delivered. [2] dead letters encountered. If this is not an expected behavior then Actor[akka://api/deadLetters] may have terminated unexpectedly. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.
[INFO] [akkaDeadLetter][01/12/2021 19:17:41.996] [api-akka.actor.default-dispatcher-7] [akka://api/deadLetters] Message [akka.actor.Status$Failure] from Actor[akka://api/user/myActor/Managemyactor#-269929265] to Actor[akka://api/deadLetters] was not delivered. [3] dead letters encountered. If this is not an expected behavior then Actor[akka://api/deadLetters] may have terminated unexpectedly. This logging can be turned off or adjusted with configuration settings 'akka.log-dead-letters' and 'akka.log-dead-letters-during-shutdown'.

请指导我哪里弄错了，或者pipeto不应该这样使用？如果是这样，我该怎么做才能让它发挥作用

解决方法

不确定它是否有意，但 ask(myManageActor,GetValue).pipeTo(sender()) 可以实现为 forward。

class MyActor extends Actor {
  lazy val myManageActor: ActorRef = ???

  override def receive: Receive = {
    case GetFinalValue =>
      myManageActor.forward(GetValue)
  }
}

forward 与 tell 相同，但它保留消息的原始发件人。

这可以应用于 MyActor 和 ManagerMyActor。

在 TokenMyActor2 的情况下，您不应该使用

future.map{ result =>
          sender ! result
      }

因为它破坏了 docs 中指定的 akka 上下文封装

当使用未来的回调，比如 onComplete，或者 map 的时候，比如 thenRun 或 thenApply 需要小心避免的内部actor 关闭包含actor的引用，即不调用方法或从内部访问封闭参与者上的可变状态打回来。这会破坏 actor 封装并可能引入同步错误和竞争条件，因为回调将同时调度给封闭的演员。不幸的是有目前还没有一种在编译时检测这些非法访问的方法。看还有：Actors 和共享可变状态

您应该使用 Future(???).pipeTo(sender())，它可以安全地与 sender() 一起使用。

应用这些更改后，代码确实按预期工作

case object GetFinalValue
case object GetValue
case object CalculateValue

class MyActor extends Actor {
  private val myManageActor: ActorRef =
    context.actorOf(Props[ManagerMyActor],"myManageActor")

  override def receive: Receive = { case GetFinalValue =>
    myManageActor.forward(GetValue)
  }
}

class ManagerMyActor extends Actor {
  private val myTokenActor =
    context.actorOf(Props[TokenMyActor2],"toknMyActor2")

  override def receive: Receive = { case GetValue =>
    myTokenActor.forward(CalculateValue)
  }

}

class TokenMyActor2 extends Actor {
  import context.dispatcher

  override def receive: Receive = { case CalculateValue =>
    val future = Future { "get the string" }
    future.pipeTo(sender())
  }
}

implicit val timeout = Timeout(3,SECONDS)
implicit val system = ActorSystem("adasd")
import system.dispatcher
val myActor = system.actorOf(Props[MyActor],"myActor")
val future = ask(myActor,GetFinalValue).mapTo[String]
future.foreach { str =>
  println(s"got $str")
}

产生got get the string。

最后一点，我建议不要在 actor 中使用 ask 模式。只需使用 ask 和 tell 即可轻松实现 forward 的基本功能。此外，代码更短，不会因不断需要 implicit val timeout

而过载 ,

只是为了添加@IvanStanislavciuc 的精彩帖子。您已经注意到您在期货中丢失了对发件人的引用。一个简单的解决方案是将其保留在前面。

这意味着在 MyActor 中发生变化：

ask(myManageActor,GetValue).pipeTo(sender()) // Won't work

进入：

val originalSender = sender()
ask(myTokenActor,CalculateValue).pipeTo(originalSender)

在ManagerMyActor中，更改：

ask(myTokenActor,CalculateValue).pipeTo(sender()) // Won't work

进入：

val originalSender = sender()
ask(myManageActor,GetValue).pipeTo(originalSender)

在TokenMyActor2中：

val originalSender = sender()
Future{ "get the string" }.pipeTo(originalSender)

代码在 Scastie 运行。

akka akka-actor scala