Mailboxes

Mailbox Selection

import akka.dispatch.BoundedMessageQueueSemantics;
import akka.dispatch.RequiresMessageQueue;

public class MyBoundedUntypedActor extends MyUntypedActor
  implements RequiresMessageQueue<BoundedMessageQueueSemantics> {
}

bounded-mailbox {
  mailbox-type = "akka.dispatch.BoundedMailbox"
  mailbox-capacity = 1000
  mailbox-push-timeout-time = 10s
}

akka.actor.mailbox.requirements {
  "akka.dispatch.BoundedMessageQueueSemantics" = bounded-mailbox
}

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

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

Builtin Mailbox Implementations

Akka comes shipped with a number of mailbox implementations:

• UnboundedMailbox (default)

  • The default mailbox
  • Backed by a java.util.concurrent.ConcurrentLinkedQueue
  • Blocking: No
  • Bounded: No
  • Configuration name: "unbounded" or "akka.dispatch.UnboundedMailbox"

• SingleConsumerOnlyUnboundedMailbox

  This queue may or may not be faster than the default one depending on your use-case—be sure to benchmark properly!

  • Backed by a Multiple-Producer Single-Consumer queue, cannot be used with BalancingDispatcher
  • Blocking: No
  • Bounded: No
  • Configuration name: "akka.dispatch.SingleConsumerOnlyUnboundedMailbox"

• NonBlockingBoundedMailbox

  • Backed by a very efficient Multiple-Producer Single-Consumer queue
  • Blocking: No (discards overflowing messages into deadLetters)
  • Bounded: Yes
  • Configuration name: "akka.dispatch.NonBlockingBoundedMailbox"

• UnboundedControlAwareMailbox

  • Delivers messages that extend akka.dispatch.ControlMessage with higher priority
  • Backed by two java.util.concurrent.ConcurrentLinkedQueue
  • Blocking: No
  • Bounded: No
  • Configuration name: "akka.dispatch.UnboundedControlAwareMailbox"

• 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: "akka.dispatch.UnboundedPriorityMailbox"

• UnboundedStablePriorityMailbox

  • Backed by a java.util.concurrent.PriorityBlockingQueue wrapped in an akka.util.PriorityQueueStabilizer
  • FIFO order is preserved for messages of equal priority - contrast with the UnboundedPriorityMailbox
  • Blocking: No
  • Bounded: No
  • Configuration name: "akka.dispatch.UnboundedStablePriorityMailbox"

Other bounded mailbox implementations which will block the sender if the capacity is reached and configured with non-zero mailbox-push-timeout-time.

• 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 "akka.dispatch.BoundedMailbox"
• BoundedPriorityMailbox
  • Backed by a java.util.PriorityQueue wrapped in an akka.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: "akka.dispatch.BoundedPriorityMailbox"
• BoundedStablePriorityMailbox
  • Backed by a java.util.PriorityQueue wrapped in an akka.util.PriorityQueueStabilizer and an akka.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: "akka.dispatch.BoundedStablePriorityMailbox"
• BoundedControlAwareMailbox
  • Delivers messages that extend akka.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: "akka.dispatch.BoundedControlAwareMailbox"

Mailbox configuration examples

public class MyPrioMailbox extends UnboundedStablePriorityMailbox {
  // needed for reflective instantiation
  public MyPrioMailbox(ActorSystem.Settings settings, Config config) {
    // Create a new PriorityGenerator, lower prio means more important
    super(new PriorityGenerator() {
      @Override
      public int gen(Object message) {
        if (message.equals("highpriority"))
          return 0; // 'highpriority messages should be treated first if possible
        else if (message.equals("lowpriority"))
          return 2; // 'lowpriority messages should be treated last if possible
        else if (message.equals(PoisonPill.getInstance()))
          return 3; // PoisonPill when no other left
        else
          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 UntypedActor {
  LoggingAdapter log = Logging.getLogger(getContext().system(), this);
  {
    for (Object msg : new Object[] { "lowpriority", "lowpriority",
        "highpriority", "pigdog", "pigdog2", "pigdog3", "highpriority",
        PoisonPill.getInstance() }) {
      getSelf().tell(msg, getSelf());
    }
  }

  public void onReceive(Object message) {
    log.info(message.toString());
  }
}

// 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(MyUntypedActor.class),
    "priomailboxactor");

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

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

public class MyControlMessage implements ControlMessage {}

class Demo extends UntypedActor {
  LoggingAdapter log = Logging.getLogger(getContext().system(), this);
  {
    for (Object msg : new Object[] { "foo", "bar", new MyControlMessage(),
            PoisonPill.getInstance() }) {
      getSelf().tell(msg, getSelf());
    }
  }

  public void onReceive(Object message) {
    log.info(message.toString());
  }
}

// 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
*/

Creating your own Mailbox type

An example is worth a thousand quacks:

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 MyUnboundedJMailbox implements MailboxType,
  ProducesMessageQueue<MyUnboundedJMailbox.MyMessageQueue> {

  // This is the MessageQueue implementation
  public static class MyMessageQueue implements MessageQueue,
    MyUnboundedJMessageQueueSemantics {
    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) {
      for (Envelope handle: queue) {
        deadLetters.enqueue(owner, handle);
      }
    }
  }

  // This constructor signature must exist, it will be called by Akka
  public MyUnboundedJMailbox(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();
  }
}

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

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.

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

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"
}

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

public class MySpecialActor extends UntypedActor implements
  RequiresMessageQueue<MyUnboundedJMessageQueueSemantics> {
  // ...
}

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,

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"));

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