Akka Typed Mailboxes 完全指南消息邮箱的选型、配置与自定义实现【免费下载链接】akka-coreA platform to build and run apps that are elastic, agile, and resilient. SDK, libraries, and hosted environments.项目地址: https://gitcode.com/gh_mirrors/ak/akka-core本指南围绕 Akka本项目 akka-coreTyped Actor API 的 Mailbox消息邮箱机制展开讲解如何在akka-actor-typed中为 Actor 选择有界/无界邮箱、通过配置路径延迟决定邮箱类型并逐一剖析 Akka 内置的十余种邮箱实现及其适用场景。阅读本文后你将掌握MailboxSelector的完整用法、邮箱配置的解析规则并能基于MailboxType编写属于自己的自定义邮箱。本文基于仓库中的官方文档 typed/mailboxes.md 编写若你使用的是 Classic Actor API可参阅对应的 Classic Mailboxes。依赖准备Mailbox 是 Akka 核心akka-actor的一部分因此在 Typed API 下使用邮箱时只需要引入akka-actor-typed依赖即可它会传递依赖核心模块。在 sbt 中按如下方式声明libraryDependencies com.typesafe.akka %% akka-actor-typed % AkkaVersionMaven 与 Gradle 用户可通过 Akka BOMakka-bom统一管理版本将AkkaVersion替换为你实际使用的版本号即可。Akka 的依赖从 Akka 官方安全仓库获取需要使用带 token 的安全 URL见 https://account.akka.io/token。邮箱是什么Actor 的消息入队缓冲区每个 Akka Actor 都拥有一个Mailbox所有发送给该 Actor 的消息在真正被 Actor 处理之前都会先进入邮箱排队。默认情况下 Akka 使用无界邮箱unbounded mailbox即允许任意数量的消息被入队。无界邮箱作为默认值是方便的但在“消息入队速度持续高于 Actor 处理速度”的场景下积压的消息会无限增长最终导致应用内存耗尽OutOfMemory。为此 Akka 提供了有界邮箱bounded mailbox当邮箱已满时新到达的消息会被转投到deadletters死信而不是无限堆积。对于更高级的用法还可以把邮箱类型的决定权延迟到配置文件代码中只指定一个绝对配置路径由运行环境决定具体使用哪种邮箱实现代码与部署解耦。为 Actor 选择邮箱类型通过 MailboxSelector 指定邮箱在 Typed API 中使用MailboxSelector来构造Props并在spawn子 Actor 时传入。以下示例同时演示了两种方式完整用例见 MailboxDocSpec.scala 与 MailboxDocTest.java 中的#select-mailbox片段Scala// 方式一直接指定有界邮箱容量为 100 context.spawn(childBehavior, bounded-mailbox-child, MailboxSelector.bounded(100)) // 方式二延迟到配置文件指向绝对配置路径 val props MailboxSelector.fromConfig(my-app.my-special-mailbox) context.spawn(childBehavior, from-config-mailbox-child, props)Java// 方式一直接指定有界邮箱容量为 100 context.spawn(childBehavior, bounded-mailbox-child, MailboxSelector.bounded(100)); // 方式二延迟到配置文件指向绝对配置路径 context.spawn( childBehavior, from-config-mailbox-child, MailboxSelector.fromConfig(my-app.my-special-mailbox));其中MailboxSelector.bounded(capacity)和MailboxSelector.fromConfig(path)的底层实现位于 Props.scalaobject MailboxSelector定义在Props.scala第 197 行附近bounded与fromConfig分别构造BoundedMailboxSelector与MailboxFromConfigSelector。fromConfig 指向的配置文件示例MailboxSelector.fromConfig(path)接收一个绝对配置路径指向配置文件中定义邮箱的那一段。示例配置见 mailbox-config-sample.confmy-app { my-special-mailbox { mailbox-type akka.dispatch.SingleConsumerOnlyUnboundedMailbox } }注意该路径是应用自定义的顶层路径如my-app.my-special-mailbox而不是嵌套在akka命名空间内部。上述配置对应的测试用例加载自ConfigFactory.load(mailbox-config-sample.conf)。默认邮箱当未显式指定邮箱时Typed API 使用默认邮箱即SingleConsumerOnlyUnboundedMailbox。配置是如何传给邮箱类型的每种邮箱类型都由一个实现了MailboxType的类承担构造函数接收两个参数ActorSystem.Settings对象一个Config配置段——它由系统从ActorSystem的配置中取出“命名配置段”计算而来计算过程中会把该配置段的id键覆盖为邮箱类型的配置路径并回退fall-back到默认邮箱配置段。这解释了为什么mailbox-type指定的类必须提供(ActorSystem.Settings, Config)构造函数——Akka 会通过反射用它来实例化邮箱。内置邮箱实现一览Akka 内置了多种邮箱实现分别适用于不同的并发与优先级场景。下表整理自 typed/mailboxes.md 的“Mailbox Implementations”一节邮箱类型底层队列阻塞有界配置名SingleConsumerOnlyUnboundedMailbox默认Multiple-Producer Single-Consumer 队列不能与BalancingDispatcher共用否否akka.dispatch.SingleConsumerOnlyUnboundedMailboxUnboundedMailboxjava.util.concurrent.ConcurrentLinkedQueue否否unbounded或akka.dispatch.UnboundedMailboxNonBlockingBoundedMailbox高效的 Multiple-Producer Single-Consumer 队列否溢出消息丢弃到 deadLetters是akka.dispatch.NonBlockingBoundedMailboxUnboundedControlAwareMailbox两个ConcurrentLinkedQueueControlMessage优先投递否否akka.dispatch.UnboundedControlAwareMailboxUnboundedPriorityMailboxjava.util.concurrent.PriorityBlockingQueue同优先级消息顺序未定义否否akka.dispatch.UnboundedPriorityMailboxUnboundedStablePriorityMailboxPriorityBlockingQueue外包一层akka.util.PriorityQueueStabilizer同优先级保持 FIFO否否akka.dispatch.UnboundedStablePriorityMailboxBoundedMailboxjava.util.concurrent.LinkedBlockingQueue配置非零mailbox-push-timeout-time时阻塞否则不阻塞是bounded或akka.dispatch.BoundedMailboxBoundedPriorityMailboxjava.util.PriorityQueue外包akka.util.BoundedBlockingQueue同优先级顺序未定义同上是akka.dispatch.BoundedPriorityMailboxBoundedStablePriorityMailboxjava.util.PriorityQueue外包PriorityQueueStabilizer与BoundedBlockingQueue同优先级保持 FIFO同上是akka.dispatch.BoundedStablePriorityMailboxBoundedControlAwareMailbox两个ConcurrentLinkedQueue容量满时入队阻塞同上是akka.dispatch.BoundedControlAwareMailbox需要特别说明的几点阻塞型有界邮箱BoundedMailbox、BoundedPriorityMailbox、BoundedStablePriorityMailbox、BoundedControlAwareMailbox只有在配置了非零mailbox-push-timeout-time时才会阻塞发送方否则它们退化为不阻塞满时消息进入 deadLetters。因此官方文档明确建议上述四个邮箱只应在mailbox-push-timeout-time为零的情况下使用。NonBlockingBoundedMailbox不使用mailbox-push-timeout-time源码注释Mailbox.scala 第 683 行附近明确指出它本质上是非阻塞的溢出消息直接进入死信。实现源码印证上述邮箱类的定义全部位于 akka-actor/src/main/scala/akka/dispatch/Mailbox.scala例如UnboundedMailbox第 648 行、SingleConsumerOnlyUnboundedMailbox第 670 行、NonBlockingBoundedMailbox第 685 行、BoundedMailbox第 700 行、UnboundedPriorityMailbox第 727 行、BoundedPriorityMailbox第 747 行、UnboundedStablePriorityMailbox第 775 行、BoundedStablePriorityMailbox第 796 行、UnboundedControlAwareMailbox第 905 行、BoundedControlAwareMailbox第 928 行。其中BoundedMailbox、BoundedPriorityMailbox、BoundedControlAwareMailbox的构造函数都从Config中读取mailbox-capacity与mailbox-push-timeout-time两个键见第 706、847、933 行附近的config.getInt(mailbox-capacity)与config.getNanosDuration(mailbox-push-timeout-time)调用这也是配置这两个参数生效的底层机制。有界邮箱的配置示例为某个 dispatcher 或邮箱配置段启用有界邮箱并设置容量与推送超时典型的 HOCON 写法如下my-bounded-dispatcher { mailbox-type akka.dispatch.BoundedMailbox mailbox-capacity 1000 # 邮箱容量 mailbox-push-timeout-time 0s # 0 表示不阻塞溢出进 deadLetters }mailbox-capacity对应构造参数capacitymailbox-push-timeout-time对应构造参数pushTimeOut均从配置段读取。自定义邮箱类型创建自定义邮箱最直接的方式是参考官方示例它由两部分组成Marker 接口/特质用于后续在配置中进行“邮箱需求映射”mailbox requirements mappingMailbox 实现类实现MailboxType与ProducesMessageQueue并提供(ActorSystem.Settings, Config)构造函数。Scala 版本Marker 特质见 MyUnboundedMailbox.scala 的#mailbox-marker-interface片段// Marker trait used for mailbox requirements mapping trait MyUnboundedMessageQueueSemantics邮箱实现见同一文件的#mailbox-implementation-example片段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 scala.Option object MyUnboundedMailbox { // This is the MessageQueue implementation class MyMessageQueue extends MessageQueue with MyUnboundedMessageQueueSemantics { private final val queue new ConcurrentLinkedQueue[Envelope]() // these should be implemented; queue used as example def enqueue(receiver: ActorRef, handle: Envelope): Unit queue.offer(handle) def dequeue(): Envelope queue.poll() def numberOfMessages: Int queue.size def hasMessages: Boolean !queue.isEmpty def cleanUp(owner: ActorRef, deadLetters: MessageQueue): Unit { while (hasMessages) { deadLetters.enqueue(owner, dequeue()) } } } } // This is the Mailbox implementation class MyUnboundedMailbox extends MailboxType with ProducesMessageQueue[MyUnboundedMailbox.MyMessageQueue] { import MyUnboundedMailbox._ // This constructor signature must exist, it will be called by Akka def this(settings: ActorSystem.Settings, config: Config) { // put your initialization code here this() } // The create method is called to create the MessageQueue final override def create(owner: Option[ActorRef], system: Option[ActorSystem]): MessageQueue new MyMessageQueue() }Java 版本Marker 接口见 MyUnboundedMessageQueueSemantics.java// Marker interface used for mailbox requirements mapping public interface MyUnboundedMessageQueueSemantics {}邮箱实现见 MyUnboundedMailbox.java 的#mailbox-implementation-example片段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.Queue; import java.util.concurrent.ConcurrentLinkedQueue; import scala.Option; public class MyUnboundedMailbox implements MailboxType, ProducesMessageQueueMyUnboundedMailbox.MyMessageQueue { // This is the MessageQueue implementation public static class MyMessageQueue implements MessageQueue, MyUnboundedMessageQueueSemantics { private final QueueEnvelope queue new ConcurrentLinkedQueueEnvelope(); // these must be implemented; queue used as example public 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) { while (!queue.isEmpty()) { deadLetters.enqueue(owner, dequeue()); } } } // This constructor signature must exist, it will be called by Akka public MyUnboundedMailbox(ActorSystem.Settings settings, Config config) { // put your initialization code here } // The create method is called to create the MessageQueue public MessageQueue create(OptionActorRef owner, OptionActorSystem system) { return new MyMessageQueue(); } }自定义邮箱的接入方式编写好上述类后将你的MailboxType的全限定类名FQCN作为mailbox-type的值写入dispatcher 配置或邮箱配置即可生效my-dispatcher { mailbox-type docs.dispatcher.MyUnboundedMailbox }自定义邮箱的关键约束必须提供(ActorSystem.Settings, Config)构造函数Akka 通过反射调用该构造函数来实例化你的邮箱类型。第二个参数Config是配置文件中描述“使用该邮箱类型的 dispatcher 或 mailbox 设置”的那一段。实例化次数邮箱类型会为每一个使用它的 dispatcher 或 mailbox 设置实例化一次。队列类型校验MessageQueue实现应实现 Marker 接口以便在需要时参与邮箱需求映射若最终创建出的队列不满足要求Actor 创建将失败此机制详见 Classic 文档 mailboxes.md 中关于RequiresMessageQueue与 dispatchermailbox-requirement的说明。更多进阶内容Classic API 下的邮箱选择优先级Classic 文档 mailboxes.md 给出了 Actor 创建时邮箱类型确定的完整顺序部署配置mailbox键 →Props中的 mailbox 选择 → dispatcher 的mailbox-type→ 邮箱需求映射 → dispatcher 需求映射 → 默认邮箱akka.actor.default-mailbox这些规则对理解 Typed API 的邮箱解析同样具有参考价值。配置为默认邮箱若希望将SingleConsumerOnlyUnboundedMailbox设为系统默认可在配置中写入akka.actor.default-mailbox { mailbox-type akka.dispatch.SingleConsumerOnlyUnboundedMailbox }。PriorityMailbox / ControlAwareMailbox 的实战配置优先级邮箱需要提供比较器Comparator控制感知邮箱需要消息实现akka.dispatch.ControlMessage完整的配置与使用示例同样收录在 Classic 文档的“Mailbox configuration examples”一节中。相关测试Typed 邮箱的文档用例位于 MailboxDocSpec.scalaScala与 MailboxDocTest.javaJava它们通过ScalaTestWithActorTestKit/TestKitJunitResource加载mailbox-config-sample.conf并实际 spawn 子 Actor 验证邮箱选择逻辑可作为你动手实验的起点。【免费下载链接】akka-coreA platform to build and run apps that are elastic, agile, and resilient. SDK, libraries, and hosted environments.项目地址: https://gitcode.com/gh_mirrors/ak/akka-core创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考