我一直在使用分布式worker模式,我遇到了一个从web请求推送工作的问题。
示例项目有一个前端:
val mediator = DistributedPubSubExtension(context.system).mediator
def receive = {
case work =>
log.info("Frontend received: " + work.toString())
implicit val timeout = Timeout(5.seconds)
(mediator ? Send("/user/master/active", work, localAffinity = false)) map {
case Master.Ack(_) => Ok
} recover { case _ => NotOk } pipeTo sender
}
和WorkProducer:
override def preStart(): Unit =
scheduler.scheduleOnce(5.seconds, self, Tick)
def receive = {
case Tick =>
n += 1
log.info("Produced work: {}", n)
val work = Work(nextWorkId(), n)
frontend ! work
context.become(waitAccepted(work), discardOld = false)
}
这一切都很好,当我从Play框架控制器直接发送到前端时:
def multiply(num: Long) = Action {
implicit request =>
implicit val timeout = Timeout(5.seconds)
val frontend = core.Main.frontend
frontend ! num
Ok
}
消息似乎丢失了。前端接收消息,但似乎下游的actor没有。
我已经修改了播放配置使用ClusterActorRefProvider
play {
akka {
extensions = ["akka.contrib.pattern.ClusterReceptionistExtension"]
actor.provider = "akka.cluster.ClusterActorRefProvider"
remote.netty.tcp.port=0
}
}
我一直是一个提线木偶,没有发送正确的消息,控制器应该是这样的:
def multiply(num: Long) = Action {
implicit request =>
implicit val timeout = Timeout(5.seconds)
val frontend = core.Main.frontend
val work = Work(nextWorkId(), num)
frontend ! work
Ok
}