首页 文章

斯卡拉阿卡演员,问模式,发送回复时遇到的死信

提问于
浏览
1

我正在尝试使用ask模式向远程actor发送请求 . 本地演员收到一些 Value ,并在其上执行一些任务并更新它 . 然后,当本地actor尝试将更新的值发送回远程actor时,发送时会发生错误 . 我该如何处理这个错误?

错误:[INFO] [03/31/2017 17:28:18.383] [ClientSystem-akka.actor.default-dispatcher-3] [akka:// ClientSystem / deadLetters]来自Actor的消息[check.package $ Rcvdcxt] [ akka:// ClientSystem / user / localA1#1050660737]到Actor [akka:// ClientSystem / deadLetters]未送达 . [1]遇到死信 .

Remote Actor:

class RemoteActor() extends Actor {

    def receive = { 


    case TaskFromLocal() =>{
        implicit val timeout: Timeout = 15000
        val currentSender = sender
        val f1 = currentSender ? RemoteActor.rtree.cxtA
            f1.onComplete{
            case Success(Rcvdcxt(cxtA))=>
                println("Success"+cxtA)
            case Success(s) =>
                    println("Success :"+s)
            case Failure(ex) =>
                    println("failure:"+ex)
            }
        }


  case _ => println("unknown msg")
  }
}
object RemoteActor{

    def createRndCxtC(count: Int):List[CxtC] = (for (i <- 1 to count) yield CxtC(Random.nextString(5), Random.nextInt())).toList
    def createRndCxtB(count: Int): List[CxtB] = (for (i <- 1 to count) yield CxtB(createRndCxtC(count), Random.nextInt())).toList
    def createRndCxtA(count: Int): List[CxtA] = (for (i <- 1 to count) yield CxtA(createRndCxtC(count), 5)).toList
    var rtree = RCxt(createRndCxtA(1),createRndCxtB(2),1,"")

    def main(args: Array[String]) {     
        val configFile = getClass.getClassLoader.getResource("remote_application.conf").getFile
        val config = ConfigFactory.parseFile(new File(configFile))
        val system = ActorSystem("RemoteSystem" , config)
        val remoteActor = system.actorOf(Props[RemoteActor], name="remote")
        println("remote is ready")
   }   
 }



Local Actor :


class LocalActorA extends Actor{    

  @throws[Exception](classOf[Exception])
  val remoteActor = context.actorSelection("akka.tcp://RemoteSystem@127.0.0.1:5150/user/remote")

  def receive = {   



    case TaskLA1(taskA) =>  {

    implicit val timeout: Timeout = 15000
      val rCxt = remoteActor ? TaskFromLocal()   
      val currentSender = sender
      rCxt.onComplete{
            case Success(Rcvdcxt(cxtA))=>
              println("Success"+cxtA)
              println("Sender: "+ sender)
              currentSender ! Rcvdcxt(cxtA)

            case Success(s)=>
              println("Got nothing from Remote"+s)
              currentSender ! "Failuree"

            case Failure(ex) =>
              println("Failure in getting remote")
              currentSender ! "Failure"
            }
    }
  }
}

object LocalActorA {    


    def createRndCxtC(count: Int):List[CxtC] = (for (i <- 1 to count) yield CxtC(Random.nextString(5), Random.nextInt())).toList
    def createRndCxtB(count: Int): List[CxtB] = (for (i <- 1 to count) yield CxtB(createRndCxtC(count), Random.nextInt())).toList
    def createRndCxtA(count: Int): List[CxtA] = (for (i <- 1 to count) yield CxtA(createRndCxtC(count), 3)).toList
    var tree = RCxt(createRndCxtA(2),createRndCxtB(2),1,"")

   def main(args: Array[String]) {
    val configFile = getClass.getClassLoader.getResource("local_application.conf").getFile
    val config = ConfigFactory.parseFile(new File(configFile))
    val system = ActorSystem("ClientSystem",config)
    val localActorA1 = system.actorOf(Props[LocalActorA], name="localA1")
    println("LocalActor A tree : "+tree)
    localActorA1 ! TaskLA1(new DummySum())
   }
}

1 回答

  • 0

    既然你没有确切地说明错误,但我最好的猜测与你在LocalActor的 onComplete 中调用 sender 的事实有关 . 这是不安全的,应该不惜一切代价避免 . 相反,做一些与远程actor类似的事情:

    class LocalActor {
      def receive = {
        case TaskLA1(taskA) =>
          val currentSender = sender
          rCxt.onComplete {
            case Success(Rcvdcxt(cxtA))=>
              currentSender ! Rcvdcxt(cxtA)
              ...
          }
      }
    }
    

相关问题