Mailboxes
Loading

Mailboxes

An Akka Mailbox holds the messages that are destined for an Actor. Normally each Actor has its own mailbox, but with for example a BalancingDispatcher all actors with the same BalancingDispatcher will share a single instance.

Mailbox Selection

Requiring a Message Queue Type for an Actor

It is possible to require a certain type of message queue for a certain type of actor by having that actor extend the parameterized trait RequiresMessageQueue. Here is an example:

  1. import akka.dispatch.RequiresMessageQueue
  2. import akka.dispatch.BoundedMessageQueueSemantics
  3.  
  4. class MyBoundedActor extends MyActor
  5.   with RequiresMessageQueue[BoundedMessageQueueSemantics]

The type parameter to the RequiresMessageQueue trait needs to be mapped to a mailbox in configuration like this:

  1. bounded-mailbox {
  2.   mailbox-type = "akka.dispatch.BoundedMailbox"
  3.   mailbox-capacity = 1000
  4.   mailbox-push-timeout-time = 10s
  5. }
  6.  
  7. akka.actor.mailbox.requirements {
  8.   "akka.dispatch.BoundedMessageQueueSemantics" = bounded-mailbox
  9. }

Now every time you create an actor of type MyBoundedActor it will try to get a bounded mailbox. If the actor has a different mailbox configured in deployment, either directly or via a dispatcher with a specified mailbox type, then that will override this mapping.

Note

The type of the queue in the mailbox created for an actor will be checked against the required type in the trait and if the queue doesn't implement the required type then actor creation will fail.

Requiring a Message Queue Type for a Dispatcher

A dispatcher may also have a requirement for the mailbox type used by the actors running on it. An example is the BalancingDispatcher which requires a message queue that is thread-safe for multiple concurrent consumers. Such a requirement is formulated within the dispatcher configuration section like this:

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

The given requirement names a class or interface which will then be ensured to be a supertype of the message queue’s implementation. In case of a conflict—e.g. if the actor requires a mailbox type which does not satisfy this requirement—then actor creation will fail.

How the Mailbox Type is Selected

When an actor is created, the ActorRefProvider first determines the dispatcher which will execute it. Then the mailbox is determined as follows:

  1. If the actor’s deployment configuration section contains a mailbox key then that names a configuration section describing the mailbox type to be used.
  2. If the actor’s Props contains a mailbox selection—i.e. withMailbox was called on it—then that names a configuration section describing the mailbox type to be used.
  3. If the dispatcher’s configuration section contains a mailbox-type key the same section will be used to configure the mailbox type.
  4. If the actor requires a mailbox type as described above then the mapping for that requirement will be used to determine the mailbox type to be used; if that fails then the dispatcher’s requirement—if any—will be tried instead.
  5. If the dispatcher requires a mailbox type as described above then the mapping for that requirement will be used to determine the mailbox type to be used.
  6. The default mailbox akka.actor.default-mailbox will be used.

Default Mailbox

When the mailbox is not specified as described above the default mailbox is used. By default it is an unbounded mailbox, which is backed by a java.util.concurrent.ConcurrentLinkedQueue.

SingleConsumerOnlyUnboundedMailbox is an even more efficient mailbox, and it can be used as the default mailbox, but it cannot be used with a BalancingDispatcher.

Configuration of SingleConsumerOnlyUnboundedMailbox as default mailbox:

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

Which Configuration is passed to the Mailbox Type

Each mailbox type is implemented by a class which extends MailboxType and takes two constructor arguments: a ActorSystem.Settings object and a Config section. The latter is computed by obtaining the named configuration section from the actor system’s configuration, overriding its id key with the configuration path of the mailbox type and adding a fall-back to the default mailbox configuration section.

Builtin implementations

Akka comes shipped with a number of mailbox implementations:

  • UnboundedMailbox - The default mailbox
    • Backed by a java.util.concurrent.ConcurrentLinkedQueue
    • Blocking: No
    • Bounded: No
    • Configuration name: "unbounded" or "akka.dispatch.UnboundedMailbox"
  • SingleConsumerOnlyUnboundedMailbox
    • Backed by a very efficient Multiple Producer Single Consumer queue, cannot be used with BalancingDispatcher
    • Blocking: No
    • Bounded: No
    • Configuration name: "akka.dispatch.SingleConsumerOnlyUnboundedMailbox"
  • BoundedMailbox
    • Backed by a java.util.concurrent.LinkedBlockingQueue
    • Blocking: Yes
    • Bounded: Yes
    • Configuration name: "bounded" or "akka.dispatch.BoundedMailbox"
  • UnboundedPriorityMailbox
    • Backed by a java.util.concurrent.PriorityBlockingQueue
    • Blocking: Yes
    • Bounded: No
    • Configuration name: "akka.dispatch.UnboundedPriorityMailbox"
  • BoundedPriorityMailbox
    • Backed by a java.util.PriorityBlockingQueue wrapped in an akka.util.BoundedBlockingQueue
    • Blocking: Yes
    • Bounded: Yes
    • Configuration name: "akka.dispatch.BoundedPriorityMailbox"
  • Durable mailboxes, see Durable Mailboxes.

Mailbox configuration examples

How to create a PriorityMailbox:

  1. import akka.dispatch.PriorityGenerator
  2. import akka.dispatch.UnboundedPriorityMailbox
  3. import com.typesafe.config.Config
  4.  
  5. // We inherit, in this case, from UnboundedPriorityMailbox
  6. // and seed it with the priority generator
  7. class MyPrioMailbox(settings: ActorSystem.Settings, config: Config)
  8.   extends UnboundedPriorityMailbox(
  9.     // Create a new PriorityGenerator, lower prio means more important
  10.     PriorityGenerator {
  11.       // 'highpriority messages should be treated first if possible
  12.       case 'highpriority ⇒ 0
  13.  
  14.       // 'lowpriority messages should be treated last if possible
  15.       case 'lowpriority  ⇒ 2
  16.  
  17.       // PoisonPill when no other left
  18.       case PoisonPill    ⇒ 3
  19.  
  20.       // We default to 1, which is in between high and low
  21.       case otherwise     ⇒ 1
  22.     })

And then add it to the configuration:

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

And then an example on how you would use it:

  1. // We create a new Actor that just prints out what it processes
  2. class Logger extends Actor {
  3.   val log: LoggingAdapter = Logging(context.system, this)
  4.  
  5.   self ! 'lowpriority
  6.   self ! 'lowpriority
  7.   self ! 'highpriority
  8.   self ! 'pigdog
  9.   self ! 'pigdog2
  10.   self ! 'pigdog3
  11.   self ! 'highpriority
  12.   self ! PoisonPill
  13.  
  14.   def receive = {
  15.     case x ⇒ log.info(x.toString)
  16.   }
  17. }
  18. val a = system.actorOf(Props(classOf[Logger], this).withDispatcher(
  19.   "prio-dispatcher"))
  20.  
  21. /*
  22.  * Logs:
  23.  * 'highpriority
  24.  * 'highpriority
  25.  * 'pigdog
  26.  * 'pigdog2
  27.  * 'pigdog3
  28.  * 'lowpriority
  29.  * 'lowpriority
  30.  */

It is also possible to configure a mailbox type directly like this:

  1. prio-mailbox {
  2.   mailbox-type = "docs.dispatcher.DispatcherDocSpec$MyPrioMailbox"
  3.   //Other mailbox configuration goes here
  4. }
  5.  
  6. akka.actor.deployment {
  7.   /priomailboxactor {
  8.     mailbox = prio-mailbox
  9.   }
  10. }

And then use it either from deployment like this:

  1. import akka.actor.Props
  2. val myActor = context.actorOf(Props[MyActor], "priomailboxactor")

Or code like this:

  1. import akka.actor.Props
  2. val myActor = context.actorOf(Props[MyActor].withMailbox("prio-mailbox"))

Creating your own Mailbox type

An example is worth a thousand quacks:

  1. import akka.actor.ActorRef
  2. import akka.actor.ActorSystem
  3. import akka.dispatch.Envelope
  4. import akka.dispatch.MailboxType
  5. import akka.dispatch.MessageQueue
  6. import akka.dispatch.ProducesMessageQueue
  7. import com.typesafe.config.Config
  8. import java.util.concurrent.ConcurrentLinkedQueue
  9. import scala.Option
  10.  
  11. // Marker trait used for mailbox requirements mapping
  12. trait MyUnboundedMessageQueueSemantics
  13.  
  14. object MyUnboundedMailbox {
  15.   // This is the MessageQueue implementation
  16.   class MyMessageQueue extends MessageQueue
  17.     with MyUnboundedMessageQueueSemantics {
  18.  
  19.     private final val queue = new ConcurrentLinkedQueue[Envelope]()
  20.  
  21.     // these must be implemented; queue used as example
  22.     def enqueue(receiver: ActorRef, handle: Envelope): Unit =
  23.       queue.offer(handle)
  24.     def dequeue(): Envelope = queue.poll()
  25.     def numberOfMessages: Int = queue.size
  26.     def hasMessages: Boolean = !queue.isEmpty
  27.     def cleanUp(owner: ActorRef, deadLetters: MessageQueue) {
  28.       while (hasMessages) {
  29.         deadLetters.enqueue(owner, dequeue())
  30.       }
  31.     }
  32.   }
  33. }
  34.  
  35. // This is the Mailbox implementation
  36. class MyUnboundedMailbox extends MailboxType
  37.   with ProducesMessageQueue[MyUnboundedMailbox.MyMessageQueue] {
  38.  
  39.   import MyUnboundedMailbox._
  40.  
  41.   // This constructor signature must exist, it will be called by Akka
  42.   def this(settings: ActorSystem.Settings, config: Config) = {
  43.     // put your initialization code here
  44.     this()
  45.   }
  46.  
  47.   // The create method is called to create the MessageQueue
  48.   final override def create(owner: Option[ActorRef],
  49.                             system: Option[ActorSystem]): MessageQueue =
  50.     new MyMessageQueue()
  51. }

And then you just specify the FQCN of your MailboxType as the value of the "mailbox-type" in the dispatcher configuration, or the mailbox configuration.

Note

Make sure to include a constructor which takes akka.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.

You can also use the mailbox as a requirement on the dispatcher like this:

  1. custom-dispatcher {
  2.   mailbox-requirement =
  3.   "docs.dispatcher.MyUnboundedJMessageQueueSemantics"
  4. }
  5.  
  6. akka.actor.mailbox.requirements {
  7.   "docs.dispatcher.MyUnboundedJMessageQueueSemantics" =
  8.   custom-dispatcher-mailbox
  9. }
  10.  
  11. custom-dispatcher-mailbox {
  12.   mailbox-type = "docs.dispatcher.MyUnboundedJMailbox"
  13. }

Or by defining the requirement on your actor class like this:

  1. class MySpecialActor extends Actor
  2.   with RequiresMessageQueue[MyUnboundedMessageQueueSemantics] {
  3.   // ...
  4. }

Special Semantics of system.actorOf

In order to make system.actorOf both synchronous and non-blocking while keeping the return type ActorRef (and the semantics that the returned ref is fully functional), special handling takes place for this case. Behind the scenes, a hollow kind of actor reference is constructed, which is sent to the system’s guardian actor who actually creates the actor and its context and puts those inside the reference. Until that has happened, messages sent to the ActorRef will be queued locally, and only upon swapping the real filling in will they be transferred into the real mailbox. Thus,

  1. val props: Props = ...
  2. // this actor uses MyCustomMailbox, which is assumed to be a singleton
  3. system.actorOf(props.withDispatcher("myCustomMailbox")) ! "bang"
  4. assert(MyCustomMailbox.instance.getLastEnqueuedMessage == "bang")

will probably fail; you will have to allow for some time to pass and retry the check à la TestKit.awaitCond.

Contents

© 2013 Typesafe Inc. Akka is Open Source and available under the Apache 2 License.

Last updated: Feb 19, 2015