9、Akka中邮箱（MailBox）

简介

Akka 的邮箱中保存着发给 Actor 的信息。通常，每个 Actor 都有自己的邮箱，但也有例外，如使用BalancingPool，则所有路由器（routees）将共享一个邮箱实例。

邮箱选择

默认邮箱

未指定邮箱时，使用默认邮箱。默认情况下，它是一个无边界的邮箱，由java.util.concurrent.ConcurrentLinkedQueue支持。

SingleConsumerOnlyUnboundedMailbox是一个效率更高的邮箱，它可以用作默认邮箱，但不能与BalancingDispatcher一起使用。

将SingleConsumerOnlyUnboundedMailbox配置为默认邮箱：

akka.actor.default-mailbox {mailbox-type = "akka.dispatch.SingleConsumerOnlyUnboundedMailbox"
}

Actor特定的邮箱

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

import akka.dispatch.BoundedMessageQueueSemantics;
import akka.dispatch.RequiresMessageQueue;public class MyBoundedActor extends MyActorimplements RequiresMessageQueue<BoundedMessageQueueSemantics> {}

RequiresMessageQueue接口的类型参数需要映射到配置中的邮箱，如下所示：

bounded-mailbox {mailbox-type = "akka.dispatch.NonBlockingBoundedMailbox"mailbox-capacity = 1000
}akka.actor.mailbox.requirements {"akka.dispatch.BoundedMessageQueueSemantics" = bounded-mailbox
}

现在，每次创建MyBoundedActor类型的 Actor 时，它都会尝试获取一个有界邮箱。如果 Actor 在部署中配置了不同的邮箱，可以直接配置，也可以通过具有指定邮箱类型的调度器（dispatcher）配置，那么这将覆盖此映射。

注释：为Actor创建的邮箱中的队列类型用接口中要求的类型进行检查，如果队列没有实现要求的类型，那么actor创建就会失败。

Dispatcher特定的邮箱

Dispatcher也需要一个邮箱类型，用于运行中的actor。一个例子就是BalancingDispatcher，它需要一个并发的、线程安全的消息队列。这样的需求可以在分发器配置中进行规划，就像这样：

my-dispatcher {mailbox-requirement = org.example.MyInterface
}

给定的需求命名了一个类或者接口，必须保证这个类或者接口是消息队列实现的超类型。万一冲突了，例如：如果actor需要一个邮箱类型，但是它不满足这个需求，那么actor创建就会失败。

如何选择邮箱类型

创建 Actor 时，ActorRefProvider首先确定将执行它的dispatcher。然后按照如下顺序确定邮箱类型：

如果actor的部署配置部分包含一个mailbox关键字，那么这个mailbox关键字就指定了要使用的邮箱类型；
如果actor的Props包含mailbox选择—即调用了withMailbox方法—那么这个方法指定要使用的邮箱类型；
如果分发器的配置部分包含一个mailbox-type关键字，那么这部分也将被用于配置邮箱类型；
如果actor需要上面描述的邮箱类型，那么这个需求的映射将被用于确定邮箱类型；如果失败了，那么分发器的需求-如果存在-将被会尝试；
如果分发器需要后面描述的邮箱类型，那么这个需求的映射将被用于确定邮箱类型；
将使用默认的邮箱akka.actor.default-mailbox。

哪些配置会传给Mailbox类型

每个邮箱类型都由一个扩展MailboxType并接受两个构造函数参数的类实现：ActorSystem.Settings对象和Config部分。后面这个是通过actor系统的配置获取，用邮箱类型的配置路径覆盖它的id关键字，并添加一个默认邮箱配置的回调。

内置邮箱实现

Akka 附带了许多邮箱实现：

UnboundedMailbox（默认）
- 默认邮箱
- 由java.util.concurrent.ConcurrentLinkedQueue支持
- 是否阻塞：No
- 是否有界：No
- 配置名称：unbounded或akka.dispatch.UnboundedMailbox
SingleConsumerOnlyUnboundedMailbox，此队列可能比默认队列快，也可能不比默认队列快，具体取决于你的用例，请确保正确地进行基准测试！
- 由多个生产商单个使用者队列支持，不能与BalancingDispatcher一起使用
- 是否阻塞：No
- 是否有界：No
- 配置名称：akka.dispatch.SingleConsumerOnlyUnboundedMailbox
NonBlockingBoundedMailbox
- 由一个非常高效的”多生产者，单消费者“队列支持
- 是否阻塞：No（将溢出的消息丢弃为deadLetters）
- 是否有界：Yes
- 配置名称：akka.dispatch.NonBlockingBoundedMailbox
UnboundedControlAwareMailbox
- 传递以更高优先级扩展akka.dispatch.ControlMessage的消息
- 由两个java.util.concurrent.ConcurrentLinkedQueue支持
- 是否阻塞：No
- 是否有界：No
- 配置名称：akka.dispatch.UnboundedControlAwareMailbox
UnboundedPriorityMailbox
- 由java.util.concurrent.PriorityBlockingQueue支持
- 等优先级邮件的传递顺序未定义，与UnboundedStablePriorityMailbox相反
- 是否阻塞：No
- 是否有界：No
- 配置名称：akka.dispatch.UnboundedPriorityMailbox
UnboundedStablePriorityMailbox
由包装在akka.util.PriorityQueueStabilizer中的java.util.concurrent.PriorityBlockingQueue提供支持
对于优先级相同的消息保留FIFO顺序，与UnboundedPriorityMailbox相反
是否阻塞：No
是否有界：No
配置名称：akka.dispatch.UnboundedStablePriorityMailbox

其他有界邮箱实现，如果达到容量并配置了非零mailbox-push-timeout-time超时时间，则会阻止发件人。特别地，以下邮箱只能与零mailbox-push-timeout-time一起使用。

BoundedMailbox
- 由java.util.concurrent.LinkedBlockingQueue支持
- 是否阻塞：如果与非零mailbox-push-timeout-time一起使用，则为Yes，否则为NO
- 是否有界：Yes
- 配置名称：bounded或akka.dispatch.BoundedMailbox
BoundedPriorityMailbox
- 由包装在akka.util.BoundedBlockingQueue中的java.util.PriorityQueue提供支持
- 优先级相同的邮件的传递顺序未定义，与BoundedStablePriorityMailbox相反
- 是否阻塞：如果与非零mailbox-push-timeout-time一起使用，则为Yes，否则为NO
- 是否有界：Yes
- 配置名称：akka.dispatch.BoundedPriorityMailbox
BoundedStablePriorityMailbox
- 由包装在akka.util.PriorityQueueStabilizer和akka.util.BoundedBlockingQueue中的java.util.PriorityQueue提供支持
- 对于优先级相同的消息保留FIFO顺序，与BoundedPriorityMailbox相反
- 是否阻塞：如果与非零mailbox-push-timeout-time一起使用，则为Yes，否则为NO
- 是否有界：Yes
- 配置名称：akka.dispatch.BoundedStablePriorityMailbox
BoundedControlAwareMailbox
- 传递以更高优先级扩展akka.dispatch.ControlMessage的消息
- 由两个java.util.concurrent.ConcurrentLinkedQueue支持，如果达到容量，则在排队时阻塞
- 是否阻塞：如果与非零mailbox-push-timeout-time一起使用，则为Yes，否则为NO
- 是否有界：Yes
- 配置名称：akka.dispatch.BoundedControlAwareMailbox

邮箱配置示例

PriorityMailbox

如何创建PriorityMailbox:

static class MyPrioMailbox extends UnboundedStablePriorityMailbox {// needed for reflective instantiationpublic MyPrioMailbox(ActorSystem.Settings settings, Config config) {// Create a new PriorityGenerator, lower prio means more importantsuper(new PriorityGenerator() {@Overridepublic int gen(Object message) {if (message.equals("highpriority"))return 0; // 'highpriority messages should be treated first if possibleelse if (message.equals("lowpriority"))return 2; // 'lowpriority messages should be treated last if possibleelse if (message.equals(PoisonPill.getInstance()))return 3; // PoisonPill when no other leftelse return 1; // By default they go between high and low prio}});}
}

然后将其添加到配置中：

prio-dispatcher {mailbox-type = "docs.dispatcher.DispatcherDocSpec$MyPrioMailbox"//Other dispatcher configuration goes here
}

下面是一个关于如何使用它的示例：

class Demo extends AbstractActor {LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);{for (Object msg :new Object[] {"lowpriority","lowpriority","highpriority","pigdog","pigdog2","pigdog3","highpriority",PoisonPill.getInstance()}) {getSelf().tell(msg, getSelf());}}@Overridepublic Receive createReceive() {return receiveBuilder().matchAny(message -> {log.info(message.toString());}).build();}
}// We create a new Actor that just prints out what it processes
ActorRef myActor =system.actorOf(Props.create(Demo.class, this).withDispatcher("prio-dispatcher"));/*
Logs:'highpriority'highpriority'pigdog'pigdog2'pigdog3'lowpriority'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(MyActor.class), "priomailboxactor");

或者这样的代码：

ActorRef myActor = system.actorOf(Props.create(MyActor.class).withMailbox("prio-mailbox"));

ControlAwareMailbox

如果 Actor 需要立即接收控制消息，无论邮箱中已经有多少其他消息，ControlAwareMailbox都非常有用。

可以这样配置：

control-aware-dispatcher {mailbox-type = "akka.dispatch.UnboundedControlAwareMailbox"//Other dispatcher configuration goes here
}

控制消息需要扩展ControlMessage特性：

static class MyControlMessage implements ControlMessage {}

下面是一个关于如何使用它的示例：

class Demo extends AbstractActor {LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);{for (Object msg :new Object[] {"foo", "bar", new MyControlMessage(), PoisonPill.getInstance()}) {getSelf().tell(msg, getSelf());}}@Overridepublic Receive createReceive() {return receiveBuilder().matchAny(message -> {log.info(message.toString());}).build();}
}// We create a new Actor that just prints out what it processes
ActorRef myActor =system.actorOf(Props.create(Demo.class, this).withDispatcher("control-aware-dispatcher"));/*
Logs:'MyControlMessage'foo'bar
*/

创建自己的邮箱类型

示例如下：

// Marker interface used for mailbox requirements mapping
public interface MyUnboundedMessageQueueSemantics {}

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.dispatch.Envelope;
import akka.dispatch.MailboxType;
import akka.dispatch.MessageQueue;
import akka.dispatch.ProducesMessageQueue;
import com.typesafe.config.Config;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.Queue;
import scala.Option;public class MyUnboundedMailboximplements MailboxType, ProducesMessageQueue<MyUnboundedMailbox.MyMessageQueue> {// This is the MessageQueue implementationpublic static class MyMessageQueue implements MessageQueue, MyUnboundedMessageQueueSemantics {private final Queue<Envelope> queue = new ConcurrentLinkedQueue<Envelope>();// these must be implemented; queue used as examplepublic void enqueue(ActorRef receiver, Envelope handle) {queue.offer(handle);}public Envelope dequeue() {return queue.poll();}public int numberOfMessages() {return queue.size();}public boolean hasMessages() {return !queue.isEmpty();}public void cleanUp(ActorRef owner, MessageQueue deadLetters) {for (Envelope handle : queue) {deadLetters.enqueue(owner, handle);}}}// This constructor signature must exist, it will be called by Akkapublic MyUnboundedMailbox(ActorSystem.Settings settings, Config config) {// put your initialization code here}// The create method is called to create the MessageQueuepublic MessageQueue create(Option<ActorRef> owner, Option<ActorSystem> system) {return new MyMessageQueue();}
}

然后，将MailboxType的 FQCN 指定为调度器配置或邮箱配置中mailbox-type的值。

注释：请确保包含一个采用akka.actor.ActorSystem.Settings和com.typesafe.config.Config参数的构造函数，因为此构造函数是通过反射调用来构造邮箱类型的。作为第二个参数传入的配置是配置中描述使用此邮箱类型的调度器或邮箱设置的部分；邮箱类型将为使用它的每个调度器或邮箱设置实例化一次。

你还可以使用邮箱作为调度器的要求（requirement），如下所示：

custom-dispatcher {mailbox-requirement ="jdocs.dispatcher.MyUnboundedMessageQueueSemantics"
}akka.actor.mailbox.requirements {"jdocs.dispatcher.MyUnboundedMessageQueueSemantics" =custom-dispatcher-mailbox
}custom-dispatcher-mailbox {mailbox-type = "jdocs.dispatcher.MyUnboundedMailbox"
}

或者像这样定义 Actor 类的要求：

static class MySpecialActor extends AbstractActorimplements RequiresMessageQueue<MyUnboundedMessageQueueSemantics> {// ...
}

system.actorOf 的特殊语义

为了使system.actorOf既同步又不阻塞，同时保持返回类型ActorRef（以及返回的ref完全起作用的语义），对这种情况进行了特殊处理。在幕后，构建了一种空的 Actor 引用，将其发送给系统的守护者 Actor，该 Actor 实际上创建了 Actor 及其上下文，并将其放入引用中。在这之前，发送到ActorRef的消息将在本地排队，只有在交换真正的填充之后，它们才会被传输到真正的邮箱中。因此，

final Props props = ...
// this actor uses MyCustomMailbox, which is assumed to be a singleton
system.actorOf(props.withDispatcher("myCustomMailbox").tell("bang", sender);
assert(MyCustomMailbox.getInstance().getLastEnqueued().equals("bang"));

可能会失败；你必须留出一段时间通过并重试检查TestKit.awaitCond。

关注公众号 数据工匠记 ，专注于大数据领域离线、实时技术干货定期分享！个人网站 www.lllpan.top