AKKA邮箱Mailbox

邮箱Mailbox

Akka Mailbox持有发给actor的消息。通常，每一个Actor都与自己的邮箱，但是当使用BalancingPool时，所有的路由都共享一个邮箱实例。

邮箱选择

Actor需要的消息队列类型

特定类型的actor可以用特定类型的消息队列，只要这个actor实现了参数化的接口RequiresMessageQueue。这里是一个例子：

如果创建PriorityMailbox：

importcom.typesafe.config.Config;

importakka.actor.ActorSystem;

importakka.actor.PoisonPill;

importakka.dispatch.PriorityGenerator;

importakka.dispatch.UnboundedPriorityMailbox;

publicclassMyPrioMailboxextendsUnboundedPriorityMailbox {

// 用于反射实例化

publicMyPrioMailbox(ActorSystem.Settingssettings, Configconfig) {

//创建一个新的PriorityGenerator，低优先级意味着更重要

super(newPriorityGenerator() {

@Override

publicintgen(Objectmessage) {

if(message.equals("highpriority"))

return0;// 如果可能的话，高优先级的消息应该被优先处理

elseif(message.equals("lowpriority"))

return2;// 如果低优先级的消息应该被最后处理

elseif(message.equals(PoisonPill.getInstance()))

return3;// 当没有剩余时，则为处理PoisonPill

else

return1;// 默认位于高优先级和低优先级

}

});

}

}

然后把它添加到配置中：

prio-dispatcher {

mailbox-type ="docs.dispatcher.DispatcherDocSpec$MyPrioMailbox"

//Other dispatcher configuration goes here

}

下面是使用这个邮箱的例子：

importcom.typesafe.config.Config;

importcom.typesafe.config.ConfigFactory;

importakka.actor.ActorRef;

importakka.actor.ActorSystem;

importakka.actor.PoisonPill;

importakka.actor.Props;

importakka.actor.UntypedActor;

importakka.event.Logging;

importakka.event.LoggingAdapter;

publicclassDemoextendsUntypedActor {

LoggingAdapterlog= Logging.getLogger(getContext().system(),this);

{

for(Objectmsg:newObject[] {"lowpriority","lowpriority","highpriority","pigdog","pigdog2","pigdog3",

"highpriority", PoisonPill.getInstance() }) {

getSelf().tell(msg, getSelf());

}

}

publicvoidonReceive(Objectmessage) {

log.info(message.toString());

}

publicstaticvoidmain(String[]args) {

Configconfig= ConfigFactory.parseString("akka.loglevel = DEBUG n"+"akka.actor.debug.lifecycle = on");

// We create a new Actor that just prints out what it processes

ActorSystemsystem= ActorSystem.create("mailbox");

ActorRefmyActor=system.actorOf(Props.create(Demo.class).withDispatcher("prio-dispatcher"),"demo");

system.terminate();

}

}

运行输出：

[INFO] [12/24/2016 23:38:22.364] [mailbox-prio-dispatcher-5] [akka://mailbox/user/demo] highpriority

[INFO] [12/24/2016 23:38:22.364] [mailbox-prio-dispatcher-5] [akka://mailbox/user/demo] highpriority

[INFO] [12/24/2016 23:38:22.364] [mailbox-prio-dispatcher-5] [akka://mailbox/user/demo] pigdog

[INFO] [12/24/2016 23:38:22.364] [mailbox-prio-dispatcher-5] [akka://mailbox/user/demo] pigdog2

[INFO] [12/24/2016 23:38:22.364] [mailbox-prio-dispatcher-5] [akka://mailbox/user/demo] pigdog3

[INFO] [12/24/2016 23:38:22.365] [mailbox-prio-dispatcher-5] [akka://mailbox/user/demo] lowpriority

[INFO] [12/24/2016 23:38:22.365] [mailbox-prio-dispatcher-5] [akka://mailbox/user/demo] lowpriority

也可以直接配置邮箱类型，就像这样：

prio-mailbox {

mailbox-type = "docs.dispatcher.DispatcherDocSpec$MyPrioMailbox"

//Other mailbox configuration goes here

}

akka.actor.deployment {

/priomailboxactor {

mailbox = prio-mailbox

}

}

然后就可以使用来自部署的邮箱类型：

ActorRef myActor =

system.actorOf(Props.create(MyUntypedActor.class),

"priomailboxactor");

或这样：

ActorRef myActor =

system.actorOf(Props.create(MyUntypedActor.class)

.withMailbox("prio-mailbox"));

一个值得上千次吹嘘的例子：

importakka.actor.ActorRef;

importakka.actor.ActorSystem;

importakka.dispatch.Envelope;

importakka.dispatch.MailboxType;

importakka.dispatch.MessageQueue;

importakka.dispatch.ProducesMessageQueue;

importcom.typesafe.config.Config;

importjava.util.concurrent.ConcurrentLinkedQueue;

importjava.util.Queue;

importscala.Option;

publicclassMyUnboundedJMailboximplementsMailboxType, ProducesMessageQueue {

// This is the MessageQueue implementation

publicstaticclassMyMessageQueueimplementsMessageQueue, MyUnboundedJMessageQueueSemantics {

privatefinalQueuequeue=newConcurrentLinkedQueue();

// these must be implemented; queue used as example

publicvoidenqueue(ActorRefreceiver, Envelopehandle) {

queue.offer(handle);

}

publicEnvelope dequeue() {

returnqueue.poll();

}

publicintnumberOfMessages() {

returnqueue.size();

}

publicbooleanhasMessages() {

return!queue.isEmpty();

}

publicvoidcleanUp(ActorRefowner, MessageQueuedeadLetters) {

for(Envelopehandle:queue) {

deadLetters.enqueue(owner,handle);

}

}

}

// This constructor signature must exist, it will be called by Akka

publicMyUnboundedJMailbox(ActorSystem.Settingssettings, Configconfig) {

// put your initialization code here

}

// The create method is called to create the MessageQueue

publicMessageQueue create(Optionowner, Optionsystem) {

returnnewMyMessageQueue();

}

}

//Marker interface used for mailbox requirements mapping

publicinterfaceMyUnboundedJMessageQueueSemantics {

}

然后只需要将分发器配置或mailbox配置的"mailbox-type"的值指定为你的MailboxType的全限定名。

注意

确保包含一个参数为akka.actor.ActorSystem.Settings和com.typesafe.config.ConfigMake的构造函数，因为这个构造函数会被反射调用，以便构建你的邮箱类型。传入的第二参数config是来自于配置中描述使用了这个邮箱类型的分发器和邮箱设置的那部分。对于每一个分发器和邮箱，邮箱类型只实例化一次。

你也可以使用邮箱类型作为分发器的必要条件，就像这样：

custom-dispatcher {

mailbox-requirement =

"docs.dispatcher.MyUnboundedJMessageQueueSemantics"

}

akka.actor.mailbox.requirements {

"docs.dispatcher.MyUnboundedJMessageQueueSemantics" =

custom-dispatcher-mailbox

}

custom-dispatcher-mailbox {

mailbox-type = "docs.dispatcher.MyUnboundedJMailbox"

}

或者在你的actor上定义必备条件，就像这样：

importakka.actor.UntypedActor;

importakka.dispatch.RequiresMessageQueue;

publicclassMySpecialActorextendsUntypedActorimplementsRequiresMessageQueue{

@Override

publicvoidonReceive(Objectarg0)throwsException {

//TODOAuto-generated method stub

}

}

为了让system.actorOf既同步又是非阻塞的，同时保持返回类型ActorRef(和返回的引用功能齐全的语义)，这种情况下需要特殊处理。在这些场景背后，构造了虚拟的actor引用，这些引用被发送到系统的守护actor，守护actor实际创建actor和它的上下文，并将它们放入引用内部。Until that has happened,发送到ActorRef的消息将被本地排队，一旦交换实际的填写，它们将被转移到真正的邮箱。因此，

finalProps props = ...

//这个actor使用MyCustomMailbox，假设它是单例

system.actorOf(props.withDispatcher("myCustomMailbox").tell("bang", sender);

assert(MyCustomMailbox.getInstance().getLastEnqueued().equals("bang"));

可能会失败；你将不得不允许一些时间来匆忙地传入检查并重试。TestKit.awaitCond。