我正在使用Akka群集(版本2.4.10),其中几个节点指定为“前端”角色,而其他几个节点则指定为“工作人员”。工人在远程机器上。前端角色通过循环路由将传入的工作分配给工人。问题是将响应从“工作人员”发送回前端参与者。我可以看到工作正在由工人完成。但是,工人发送给前端的消息没有到达并最终变成死信。我在日志中看到以下错误。
[Cluster-akka.actor.default-dispatcher-21] [akka://Cluster/deadLetters] Message [scala.collection.immutable.$colon$colon] from Actor[akka://Cluster/user] to Actor[akka://Cluster/deadLetters] was not delivered. [6] dead letters encountered.
我已经看到了这一点,并且在我的代码中也遵循了同样的观点。我也看到了这一点,但是建议的解决方案在这种情况下不适用,因为我不知道预先的路线。它来自配置,并且可以更改。循环路由器的配置如下。
akka.actor.deployment {
/frontEnd/hm = {
router = round-robin-group
nr-of-instances = 5
routees.paths = ["/user/hmWorker"]
cluster {
enabled = on
use-role = backend
allow-local-routees = on
}
}
}
路由器在前端参与者中实例化,如下所示。
val router = context.actorOf(FromConfig.props(), name = "hm")
val controller = context.actorOf(Props(classOf[Controller], router))
控制器和工作人员代码如下。
// Node 1 : Controller routes requests using round-robin
class Controller(router: ActorRef) extends Actor {
val list = List("a", "b") // Assume this is a big list
val groups = list.grouped(500)
override def receive: Actor.Receive = {
val futures = groups.map(grp => (router ? Message(grp)).mapTo[List[String]]))
val future = Future.sequence(futures).map(_.flatten)
val result = Await.result(future, 50 seconds)
println(s"Result is $result")
}
}
// Node 2
class Worker extends Actor {
override def receive: Actor.Receive = {
case Message(lst) =>
val future: Future[List[String]] = // Do Something asynchronous
future onComplete {
case Success(r) => sender.!(r)(context.parent) // This message is not delivered to Controller actor.
case Failure(th) => // Error handling
}
}
}
请让我知道我在做什么错。感谢您的帮助。
您不应该sender()
在的回调中使用Future
。在处理回调时,sender()
可能指的是与您收到消息时不同的东西。
考虑先将引用保存在回调之外,例如:
override def receive: Actor.Receive = {
case Message(lst) =>
val future: Future[List[String]] = // Do Something asynchronous
val replyTo: ActorRef = sender()
future onComplete {
case Success(r) => replyTo.!(r)(context.parent) // This message is not delivered to Controller actor.
case Failure(th) => // Error handling
}
}
甚至更好的是,使用管道模式:
import akka.pattern.pipe
override def receive: Actor.Receive = {
case Message(lst) =>
val future: Future[List[String]] = // Do Something asynchronous
future.pipeTo(sender())
}
本文收集自互联网,转载请注明来源。
如有侵权,请联系[email protected] 删除。
我来说两句