Dispatchers (Scala)
An Akka MessageDispatcher is what makes Akka Actors "tick", it is the engine of the machine so to speak. All MessageDispatcher implementations are also an ExecutionContext, which means that they can be used to execute arbitrary code, for instance Futures (Scala).
Default dispatcher
Every ActorSystem will have a default dispatcher that will be used in case nothing else is configured for an Actor. The default dispatcher can be configured, and is by default a Dispatcher with a "fork-join-executor", which gives excellent performance in most cases.
Setting the dispatcher for an Actor
So in case you want to give your Actor a different dispatcher than the default, you need to do two things, of which the first is:
- import akka.actor.Props
- val myActor =
- context.actorOf(Props[MyActor].withDispatcher("my-dispatcher"), "myactor1")
Note
The "dispatcherId" you specify in withDispatcher is in fact a path into your configuration. So in this example it's a top-level section, but you could for instance put it as a sub-section, where you'd use periods to denote sub-sections, like this: "foo.bar.my-dispatcher"
And then you just need to configure that dispatcher in your configuration:
- my-dispatcher {
- # Dispatcher is the name of the event-based dispatcher
- type = Dispatcher
- # What kind of ExecutionService to use
- executor = "fork-join-executor"
- # Configuration for the fork join pool
- fork-join-executor {
- # Min number of threads to cap factor-based parallelism number to
- parallelism-min = 2
- # Parallelism (threads) ... ceil(available processors * factor)
- parallelism-factor = 2.0
- # Max number of threads to cap factor-based parallelism number to
- parallelism-max = 10
- }
- # Throughput defines the maximum number of messages to be
- # processed per actor before the thread jumps to the next actor.
- # Set to 1 for as fair as possible.
- throughput = 100
- }
And here's another example that uses the "thread-pool-executor":
- my-thread-pool-dispatcher {
- # Dispatcher is the name of the event-based dispatcher
- type = Dispatcher
- # What kind of ExecutionService to use
- executor = "thread-pool-executor"
- # Configuration for the thread pool
- thread-pool-executor {
- # minimum number of threads to cap factor-based core number to
- core-pool-size-min = 2
- # No of core threads ... ceil(available processors * factor)
- core-pool-size-factor = 2.0
- # maximum number of threads to cap factor-based number to
- core-pool-size-max = 10
- }
- # Throughput defines the maximum number of messages to be
- # processed per actor before the thread jumps to the next actor.
- # Set to 1 for as fair as possible.
- throughput = 100
- }
For more options, see the default-dispatcher section of the Configuration.
Types of dispatchers
There are 4 different types of message dispatchers:
Dispatcher
This is an event-based dispatcher that binds a set of Actors to a thread pool. It is the default dispatcher used if one is not specified.
Sharability: Unlimited
Mailboxes: Any, creates one per Actor
Use cases: Default dispatcher, Bulkheading
- Driven by: java.util.concurrent.ExecutorService
specify using "executor" using "fork-join-executor", "thread-pool-executor" or the FQCN of an akka.dispatcher.ExecutorServiceConfigurator
PinnedDispatcher
This dispatcher dedicates a unique thread for each actor using it; i.e. each actor will have its own thread pool with only one thread in the pool.
Sharability: None
Mailboxes: Any, creates one per Actor
Use cases: Bulkheading
- Driven by: Any akka.dispatch.ThreadPoolExecutorConfigurator
by default a "thread-pool-executor"
BalancingDispatcher
This is an executor based event driven dispatcher that will try to redistribute work from busy actors to idle actors.
All the actors share a single Mailbox that they get their messages from.
It is assumed that all actors using the same instance of this dispatcher can process all messages that have been sent to one of the actors; i.e. the actors belong to a pool of actors, and to the client there is no guarantee about which actor instance actually processes a given message.
Sharability: Actors of the same type only
Mailboxes: Any, creates one for all Actors
Use cases: Work-sharing
- Driven by: java.util.concurrent.ExecutorService
specify using "executor" using "fork-join-executor", "thread-pool-executor" or the FQCN of an akka.dispatcher.ExecutorServiceConfigurator
Note that you can not use a BalancingDispatcher as a Router Dispatcher. (You can however use it for the Routees)
CallingThreadDispatcher
- This dispatcher runs invocations on the current thread only. This dispatcher does not create any new threads, but it can be used from different threads concurrently for the same actor. See CallingThreadDispatcher for details and restrictions.
- Sharability: Unlimited
- Mailboxes: Any, creates one per Actor per Thread (on demand)
- Use cases: Testing
- Driven by: The calling thread (duh)
More dispatcher configuration examples
Configuring a PinnedDispatcher:
- my-pinned-dispatcher {
- executor = "thread-pool-executor"
- type = PinnedDispatcher
- }
And then using it:
- val myActor =
- context.actorOf(Props[MyActor].withDispatcher("my-pinned-dispatcher"), "myactor2")
Note that thread-pool-executor configuration as per the above my-thread-pool-dispatcher example is NOT applicable. This is because every actor will have its own thread pool when using PinnedDispatcher,
Mailboxes
An Akka Mailbox holds the messages that are destined for an Actor. Normally each Actor has its own mailbox, but with example a BalancingDispatcher all actors with the same BalancingDispatcher will share a single instance.
Builtin implementations
Akka comes shipped with a number of default mailbox implementations:
- UnboundedMailbox
- Backed by a java.util.concurrent.ConcurrentLinkedQueue
- Blocking: No
- Bounded: No
- BoundedMailbox
- Backed by a java.util.concurrent.LinkedBlockingQueue
- Blocking: Yes
- Bounded: Yes
- UnboundedPriorityMailbox
- Backed by a java.util.concurrent.PriorityBlockingQueue
- Blocking: Yes
- Bounded: No
- BoundedPriorityMailbox
- Backed by a java.util.PriorityBlockingQueue wrapped in an akka.util.BoundedBlockingQueue
- Blocking: Yes
- Bounded: Yes
- Durable mailboxes, see Durable Mailboxes.
Mailbox configuration examples
How to create a PriorityMailbox:
- import akka.dispatch.PriorityGenerator
- import akka.dispatch.UnboundedPriorityMailbox
- import com.typesafe.config.Config
-
- // We inherit, in this case, from UnboundedPriorityMailbox
- // and seed it with the priority generator
- class MyPrioMailbox(settings: ActorSystem.Settings, config: Config) extends UnboundedPriorityMailbox(
- // Create a new PriorityGenerator, lower prio means more important
- PriorityGenerator {
- // 'highpriority messages should be treated first if possible
- case 'highpriority ⇒ 0
-
- // 'lowpriority messages should be treated last if possible
- case 'lowpriority ⇒ 2
-
- // PoisonPill when no other left
- case PoisonPill ⇒ 3
-
- // We default to 1, which is in between high and low
- case otherwise ⇒ 1
- })
And then add it to the configuration:
- prio-dispatcher {
- mailbox-type = "akka.docs.dispatcher.DispatcherDocSpec$MyPrioMailbox"
- }
And then an example on how you would use it:
- // We create a new Actor that just prints out what it processes
- val a = system.actorOf(
- Props(new Actor {
- val log: LoggingAdapter = Logging(context.system, this)
-
- self ! 'lowpriority
- self ! 'lowpriority
- self ! 'highpriority
- self ! 'pigdog
- self ! 'pigdog2
- self ! 'pigdog3
- self ! 'highpriority
- self ! PoisonPill
-
- def receive = {
- case x ⇒ log.info(x.toString)
- }
- }).withDispatcher("prio-dispatcher"))
-
- /*
- Logs:
- 'highpriority
- 'highpriority
- 'pigdog
- 'pigdog2
- 'pigdog3
- 'lowpriority
- 'lowpriority
- */
Creating your own Mailbox type
An example is worth a thousand quacks:
- case class MyUnboundedMailbox() extends akka.dispatch.MailboxType {
- import akka.actor.ActorContext
- import com.typesafe.config.Config
- import java.util.concurrent.ConcurrentLinkedQueue
- import akka.dispatch.{
- Envelope,
- MessageQueue,
- QueueBasedMessageQueue,
- UnboundedMessageQueueSemantics
- }
-
- // This constructor signature must exist, it will be called by Akka
- def this(settings: ActorSystem.Settings, config: Config) = this()
-
- // The create method is called to create the MessageQueue
- final override def create(owner: Option[ActorContext]): MessageQueue =
- new QueueBasedMessageQueue with UnboundedMessageQueueSemantics {
- final val queue = new ConcurrentLinkedQueue[Envelope]()
- }
And then you just specify the FQCN of your MailboxType as the value of the "mailbox-type" in the dispatcher 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 using this mailbox type; the mailbox type will be instantiated once for each dispatcher using it.