1

实际上,我无法让我的演员(路由器)系统正常工作。我的设置:

我正在尝试在播放控制器中使用 akka 路由器。对于依赖注入,我使用 scaldi。

斯卡迪模块:

class UserDAOModule extends Module {
  binding to new ExampleRouter
  binding toProvider new UserDAOWorker
}

akka路由器:

class UserDAORouter(implicit inj:Injector) extends Actor with AkkaInjectable {

  val userDAOProps = injectActorProps[UserDAOWorker]

  var router = {
    val routees = Vector.fill(5) {
      val r = context.actorOf(userDAOProps)
      context watch r
      ActorRefRoutee(r)
    }
    Router(RoundRobinRoutingLogic(), routees)
  }

  override def receive: Receive = {
    case mm: MongoDBMessage =>
      router.route(mm, sender)
    case Terminated(a) =>
      router = router.removeRoutee(a)
      val r = context.actorOf(userDAOProps)
      context watch r
      router = router.addRoutee(r)
  }

}

工人:

class UserDAOWorker(implicit inj:Injector) extends Actor with Injectable {

  val db = inject[DefaultDB]
  val collection:JSONCollection = db("users")
  val currentSender = sender

  override def receive: Receive = {
    case InsertUser(user) => insertUser(user)
  }

  def insertUser(user:User) = {
    collection.save(user).onComplete {
      case Failure(e) => currentSender ! new UserDAOReturnMessage(Some(e), None)
      case Success(lastError) => currentSender ! new UserDAOReturnMessage(None, lastError)
    }
  }
}

当我向路由器发送消息(插入用户消息)时,它被正确路由并且工作人员接收到消息,但是当工作人员将消息发送回发件人时,它无法传递,所以它被发送到死信办公室。我不知道如何解决这个问题。有人可以帮助我吗?

提前致谢

4

1 回答 1

1

我猜问题是 currentSender 在创建 actor 的构造函数中被初始化为 null(即 ActorRef.noSender)。'sender' 仅在在 receive() 中接收消息的上下文中有效。向 ActorRef.noSender 发送消息相当于向死信队列发送消息。

像这样的东西应该工作:

class UserDAOWorker(implicit inj:Injector) extends Actor with Injectable {

  val db = inject[DefaultDB]
  val collection:JSONCollection = db("users")

  override def receive: Receive = {
    case InsertUser(user) => {
      insertUser(sender, user)
    }
  }

  def insertUser(currentSender : ActorRef, user:User) = {
    collection.save(user).onComplete {
      case Failure(e) => currentSender ! new UserDAOReturnMessage(Some(e), None)
      case Success(lastError) => currentSender ! new UserDAOReturnMessage(None, lastError)
    }
  }
}
于 2014-09-26T22:47:13.400 回答