Mailboxes
You are viewing the documentation for the new actor APIs, to view the Pekko Classic documentation, see Classic Mailboxes.
Dependency
Mailboxes are part of core Pekko, which means that they are part of the pekko-actor
dependency. This page describes how to use mailboxes with pekko-actor-typed
, which has dependency:
- sbt
val PekkoVersion = "1.1.1" libraryDependencies += "org.apache.pekko" %% "pekko-actor-typed" % PekkoVersion
- Maven
<properties> <scala.binary.version>2.13</scala.binary.version> </properties> <dependencyManagement> <dependencies> <dependency> <groupId>org.apache.pekko</groupId> <artifactId>pekko-bom_${scala.binary.version}</artifactId> <version>1.1.1</version> <type>pom</type> <scope>import</scope> </dependency> </dependencies> </dependencyManagement> <dependencies> <dependency> <groupId>org.apache.pekko</groupId> <artifactId>pekko-actor-typed_${scala.binary.version}</artifactId> </dependency> </dependencies>
- Gradle
def versions = [ ScalaBinary: "2.13" ] dependencies { implementation platform("org.apache.pekko:pekko-bom_${versions.ScalaBinary}:1.1.1") implementation "org.apache.pekko:pekko-actor-typed_${versions.ScalaBinary}" }
Introduction
Each actor in Pekko has a Mailbox
, this is where the messages are enqueued before being processed by the actor.
By default an unbounded mailbox is used, this means any number of messages can be enqueued into the mailbox.
The unbounded mailbox is a convenient default but in a scenario where messages are added to the mailbox faster than the actor can process them, this can lead to the application running out of memory. For this reason a bounded mailbox can be specified, the bounded mailbox will pass new messages to deadletters
when the mailbox is full.
For advanced use cases it is also possible to defer mailbox selection to config by pointing to a config path.
Selecting what mailbox is used
Selecting a Mailbox Type for an Actor
To select a specific mailbox for an actor use MailboxSelector
MailboxSelector
to create a Props
Props
instance for spawning your actor:
- Scala
-
source
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
-
source
context.spawn(childBehavior, "bounded-mailbox-child", MailboxSelector.bounded(100)); context.spawn( childBehavior, "from-config-mailbox-child", MailboxSelector.fromConfig("my-app.my-special-mailbox"));
fromConfig
fromConfig
takes an absolute config path to a block defining the dispatcher in the config file:
source# SPDX-License-Identifier: Apache-2.0
my-app {
my-special-mailbox {
mailbox-type = "org.apache.pekko.dispatch.SingleConsumerOnlyUnboundedMailbox"
}
}
Default Mailbox
The default mailbox is used when the mailbox is not specified and is the SingleConsumerOnlyUnboundedMailbox
SingleConsumerOnlyUnboundedMailbox
Which Configuration is passed to the Mailbox Type
Each mailbox type is implemented by a class which extends MailboxType
MailboxType
and takes two constructor arguments: a ActorSystem.Settings
ActorSystem.Settings
object and a Config section. The latter is computed by obtaining the named configuration section from the ActorSystem
ActorSystem
configuration, overriding its id
key with the configuration path of the mailbox type and adding a fall-back to the default mailbox configuration section.
Interoperability with DispatcherSelector
The MailboxSelector
MailboxSelector
will create a Props
Props
instance that can be both set up Dispatcher and Mailbox, which means that you can continue to set up Dispatcher through chain calls.
- Scala
-
source
context.spawn(childBehavior, "bounded-mailbox-child", MailboxSelector.bounded(100).withDispatcherDefault) val props = MailboxSelector.fromConfig("my-app.my-special-mailbox").withDispatcherFromConfig( Dispatchers.DefaultDispatcherId) context.spawn(childBehavior, "from-config-mailbox-child", props)
- Java
-
source
context.spawn( childBehavior, "bounded-mailbox-child", MailboxSelector.bounded(100).withDispatcherDefault()); context.spawn( childBehavior, "from-config-mailbox-child", MailboxSelector.fromConfig("my-app.my-special-mailbox") .withDispatcherFromConfig(Dispatchers.DefaultDispatcherId()));
Mailbox Implementations
Pekko ships with a number of mailbox implementations:
-
SingleConsumerOnlyUnboundedMailbox
SingleConsumerOnlyUnboundedMailbox
(default)- This is the default
- Backed by a Multiple-Producer Single-Consumer queue, cannot be used with
BalancingDispatcher
- Blocking: No
- Bounded: No
- Configuration name:
"org.apache.pekko.dispatch.SingleConsumerOnlyUnboundedMailbox"
-
UnboundedMailbox
UnboundedMailbox
- Backed by a
java.util.concurrent.ConcurrentLinkedQueue
- Blocking: No
- Bounded: No
- Configuration name:
"unbounded"
or"org.apache.pekko.dispatch.UnboundedMailbox"
- Backed by a
-
NonBlockingBoundedMailbox
NonBlockingBoundedMailbox
- Backed by a very efficient Multiple-Producer Single-Consumer queue
- Blocking: No (discards overflowing messages into deadLetters)
- Bounded: Yes
- Configuration name:
"org.apache.pekko.dispatch.NonBlockingBoundedMailbox"
-
UnboundedControlAwareMailbox
UnboundedControlAwareMailbox
- Delivers messages that extend
dispatch.ControlMessage
dispatch.ControlMessage
with higher priority - Backed by two
java.util.concurrent.ConcurrentLinkedQueue
- Blocking: No
- Bounded: No
- Configuration name:
"org.apache.pekko.dispatch.UnboundedControlAwareMailbox"
- Delivers messages that extend
-
UnboundedPriorityMailbox
UnboundedPriorityMailbox
- Backed by a
java.util.concurrent.PriorityBlockingQueue
- Delivery order for messages of equal priority is undefined - contrast with the
UnboundedStablePriorityMailbox
- Blocking: No
- Bounded: No
- Configuration name:
"org.apache.pekko.dispatch.UnboundedPriorityMailbox"
- Backed by a
-
UnboundedStablePriorityMailbox
UnboundedStablePriorityMailbox
- Backed by a
java.util.concurrent.PriorityBlockingQueue
wrapped in anutil.PriorityQueueStabilizer
util.PriorityQueueStabilizer
- FIFO order is preserved for messages of equal priority - contrast with the
UnboundedPriorityMailbox
- Blocking: No
- Bounded: No
- Configuration name:
"org.apache.pekko.dispatch.UnboundedStablePriorityMailbox"
- Backed by a
Other bounded mailbox implementations which will block the sender if the capacity is reached and configured with non-zero mailbox-push-timeout-time
.
The following mailboxes should only be used with zero mailbox-push-timeout-time
.
BoundedMailbox
BoundedMailbox
- Backed by a
java.util.concurrent.LinkedBlockingQueue
- Blocking: Yes if used with non-zero
mailbox-push-timeout-time
, otherwise No - Bounded: Yes
- Configuration name:
"bounded"
or"org.apache.pekko.dispatch.BoundedMailbox"
- Backed by a
BoundedPriorityMailbox
BoundedPriorityMailbox
- Backed by a
java.util.PriorityQueue
wrapped in anutil.BoundedBlockingQueue
util.BoundedBlockingQueue
- Delivery order for messages of equal priority is undefined - contrast with the
BoundedStablePriorityMailbox
- Blocking: Yes if used with non-zero
mailbox-push-timeout-time
, otherwise No - Bounded: Yes
- Configuration name:
"org.apache.pekko.dispatch.BoundedPriorityMailbox"
- Backed by a
BoundedStablePriorityMailbox
BoundedStablePriorityMailbox
- Backed by a
java.util.PriorityQueue
wrapped in anutil.PriorityQueueStabilizer
util.PriorityQueueStabilizer
and anutil.BoundedBlockingQueue
util.BoundedBlockingQueue
- FIFO order is preserved for messages of equal priority - contrast with the BoundedPriorityMailbox
- Blocking: Yes if used with non-zero
mailbox-push-timeout-time
, otherwise No - Bounded: Yes
- Configuration name:
"org.apache.pekko.dispatch.BoundedStablePriorityMailbox"
- Backed by a
BoundedControlAwareMailbox
BoundedControlAwareMailbox
- Delivers messages that extend
dispatch.ControlMessage
dispatch.ControlMessage
with higher priority - Backed by two
java.util.concurrent.ConcurrentLinkedQueue
and blocking on enqueue if capacity has been reached - Blocking: Yes if used with non-zero
mailbox-push-timeout-time
, otherwise No - Bounded: Yes
- Configuration name:
"org.apache.pekko.dispatch.BoundedControlAwareMailbox"
- Delivers messages that extend
Custom Mailbox type
The best way to show how to create your own Mailbox type is by example:
- Scala
-
source
// Marker trait used for mailbox requirements mapping trait MyUnboundedMessageQueueSemantics
- Java
-
source
// Marker interface used for mailbox requirements mapping public interface MyUnboundedMessageQueueSemantics {}
- Scala
-
source
import org.apache.pekko import pekko.actor.ActorRef import pekko.actor.ActorSystem import pekko.dispatch.Envelope import pekko.dispatch.MailboxType import pekko.dispatch.MessageQueue import pekko.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 Pekko 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
-
source
import org.apache.pekko.actor.ActorRef; import org.apache.pekko.actor.ActorSystem; import org.apache.pekko.dispatch.Envelope; import org.apache.pekko.dispatch.MailboxType; import org.apache.pekko.dispatch.MessageQueue; import org.apache.pekko.dispatch.ProducesMessageQueue; import com.typesafe.config.Config; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.Queue; import scala.Option; public class MyUnboundedMailbox implements MailboxType, ProducesMessageQueue<MyUnboundedMailbox.MyMessageQueue> { // This is the MessageQueue implementation public static class MyMessageQueue implements MessageQueue, MyUnboundedMessageQueueSemantics { private final Queue<Envelope> queue = new ConcurrentLinkedQueue<Envelope>(); // 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 Pekko public MyUnboundedMailbox(ActorSystem.Settings settings, Config config) { // put your initialization code here } // The create method is called to create the MessageQueue public MessageQueue create(Option<ActorRef> owner, Option<ActorSystem> system) { return new MyMessageQueue(); } }
And then you specify the FQCN of your MailboxType as the value of the “mailbox-type” in the dispatcher configuration, or the mailbox configuration.
Make sure to include a constructor which takes actor.ActorSystem.Settings
actor.ActorSystem.Settings
and com.typesafe.config.Config arguments, as this constructor is invoked reflectively to construct your mailbox type. The config passed in as second argument is that section from the configuration which describes the dispatcher or mailbox setting using this mailbox type; the mailbox type will be instantiated once for each dispatcher or mailbox setting using it.