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