Dispatchers (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.

Looking up a Dispatcher

Dispatchers implement the ExecutionContext interface and can thus be used to run Future invocations etc.

// for use with Futures, Scheduler, etc.
implicit val executionContext = system.dispatchers.lookup("my-dispatcher")

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

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)

my-pinned-dispatcher {
  executor = "thread-pool-executor"
  type = PinnedDispatcher
}

val myActor =
  context.actorOf(Props[MyActor].withDispatcher("my-pinned-dispatcher"), "myactor2")

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.

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

prio-dispatcher {
  mailbox-type = "docs.dispatcher.DispatcherDocSpec$MyPrioMailbox"
}

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

class MyUnboundedMailbox extends akka.dispatch.MailboxType {
  import akka.actor.{ ActorRef, ActorSystem }
  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[ActorRef],
                            system: Option[ActorSystem]): MessageQueue =
    new QueueBasedMessageQueue with UnboundedMessageQueueSemantics {
      final val queue = new ConcurrentLinkedQueue[Envelope]()
    }
}

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