【发布时间】:2021-06-06 20:38:36
【问题描述】:
我正在尝试使用 akka 类型创建一个自定义优先级邮箱,该邮箱根据消息类型对消息进行优先级排序。以下是我创建演员并向其发送消息的代码。但是,消息是按照接收顺序而不是按照自定义邮箱定义的顺序处理的。
package com.akka.prac
import akka.actor.typed.{ActorSystem, Behavior, DispatcherSelector, Props, Settings}
import akka.actor.typed.scaladsl.Behaviors
import akka.dispatch.{PriorityGenerator, UnboundedPriorityMailbox}
import com.typesafe.config.{Config, ConfigFactory}
object PriorityMailBoxTyped extends App {
val config = ConfigFactory.load("applicationtyped.conf")
val actorSystem = ActorSystem(
MyPriorityActorTyped.receive,
"PriorityMailBox",
config,
DispatcherSelector.fromConfig("prio-dispatcher")
)
actorSystem ! DoubleMessage(6.0)
actorSystem ! IntMessage(1)
actorSystem ! DoubleMessage(5.0)
actorSystem ! IntMessage(3)
actorSystem ! StringMessage("Hello")
actorSystem ! IntMessage(5)
actorSystem ! StringMessage("I am priority actor")
actorSystem ! StringMessage(
"I process string messages first,then integer, long and others"
)
}
class MyPriorityActorMailBoxTyped(settings: Settings, config: Config)
extends UnboundedPriorityMailbox(
// Create a new PriorityGenerator,lower prio means more important
PriorityGenerator {
// Int Messages
case x: IntMessage => 1
// String Messages
case x: StringMessage => 0
// Long messages
case x: LongMessage => 2
// other messages
case x: DoubleMessage => 3
case _ => 4
}
)
trait MyMessage
case class IntMessage(x: Int) extends MyMessage
case class LongMessage(x: Long) extends MyMessage
case class StringMessage(x: String) extends MyMessage
case class DoubleMessage(x: Double) extends MyMessage
object MyPriorityActorTyped {
def receive: Behavior[MyMessage] = Behaviors.receive { (context, message) =>
message match {
case IntMessage(x) => println(x)
case StringMessage(x) => println(x)
case LongMessage(x) => println(x)
case DoubleMessage(x) => println(x)
case _ => println()
}
Behaviors.same
}
}
配置文件
prio-dispatcher {
mailbox-type = "com.akka.prac.MyPriorityActorMailBoxTyped"
}
同样的例子也适用于 Akka Classic。我在这里错过了什么?
谢谢。
【问题讨论】:
-
我想这是因为消息处理得太快了,所以 ActorSystem 没有改变消息的优先级。尝试添加睡眠以接收。
-
谢谢@IvanKurchenko。我尝试通过添加 1 秒睡眠来接收但得到了相同的结果。 \n 更新的 Actor: 'object MyPriorityActorTyped { def receive: Behavior[MyMessage] = Behaviors.receive{ (context,message) => Thread.sleep(1000) message match { case IntMessage(x) => println(x) case StringMessage (x) => println(x) case LongMessage(x) => println(x) case DoubleMessage(x) => println(x) case _ => println() } Behaviors.same } } ` 输出:6.0 1 5.0 3 你好 5 我是优先演员我处理字符串消息...
-
你在创建actor时是否使用了MailboxSelector?似乎是新的 Typed 方式:doc.akka.io/docs/akka/current/typed/mailboxes.html
-
感谢@FlorianSchaetz。我将代码中的 DispatcherSelector 更改为 MailboxSelector。不幸的是同样的结果!
标签: scala akka akka-typed