Using Akka Dispatchers In Scala

Table of contents

Reading Time: 4 minutes

Akka dispatchers are extremely important in Akka framework. They are directly responsible for optimal performance, throughput and scalability.

Akka supports dispatchers for both event-driven lightweight threads and thread-based Actors. For thread-based Actors each dispatcher is bound to a dedicated OS thread.

Default dispatcher is a single event-based dispatcher for all Actors created. The dispatcher used is this one:

Dispatchers.globalExecutorBasedEventDrivenDispatcher

For many cases it becomes mandatory to group Actors together for a dedicated dispatcher, then we can override the defaults and define our own dispatcher.

Setting the Dispatcher
Normally we set the dispatcher in Actor itself

class EchoActor extends Actor {
     self.dispatcher = ..... // set the dispatcher
}

Or we can set it in the ActorRef

actorRef.dispatcher = dispatcher

There are different kind of dispatchers

Thread-based
Event-based
Priority event-based
Work-stealing

Thread-based
It binds dedicated OS thread to each Actor. The messages are posted to LinkedBlockingQueue which feeds messages to dispatcher one by one. It has worst performance and scalability. We also cannot share it among actors. Although Actors do not block for threads in this case.
Code example

class EchoActor extends Actor {
     self.dispatcher = Dispatchers.newThreadBasedDispatcher(self)
     ....
}

Event-based
The ExecutorBasedEventDrivenDispatcher binds a set of Actors to a thread pool backed up by a BlockingQueue. The dispatcher must be shared among Actors. This dispatcher is highly configurable and here we can specify things like ‘type of queue’, ‘max items’ , ‘rejection-policy’.
Code example

class EchoActor extends Actor {
self.dispatcher = Dispatchers.newExecutorBasedEventDrivenDispatcher(name)
 .withNewThreadPoolWithLinkedBlockingQueueWithCapacity(100)
 .setCorePoolSize(16)
 .setMaxPoolSize(128)
 .setKeepAliveTimeInMillis(60000)
 .setRejectionPolicy(new CallerRunsPolicy)
 .build
 ....
}

Priority event-based
It is meant for handling messages when priorities are assigned to messages. It is done by using PriorityExecutorBasedEventDrivenDispatcher. It requires a PriorityGenerator as an attribute in its constructor.

Let’s look at an example where we have a PriorityExecutorBasedEventDrivenDispatcher used for a group of messages fired on an actor.

package com.meetu.akka.dispatcher

import akka.actor.Actor.actorOf
import akka.actor.Actor
import akka.dispatch.PriorityExecutorBasedEventDrivenDispatcher
import akka.dispatch.PriorityGenerator

object PriorityDispatcherExample extends App {
  val actor = Actor.actorOf(
    new Actor {
      def receive = {
        case x => println(x)
      }
    })

  val priority = PriorityGenerator {
    case "high priority" => 0
    case "low priority" => 100
    case _ => 50
  }

  actor.dispatcher = new PriorityExecutorBasedEventDrivenDispatcher("foo", priority)

  actor.start
  actor.dispatcher.suspend(actor)

  actor ! "low priority"
  actor ! "others"
  actor ! "low priority"
  actor ! "high priority"

  actor.dispatcher.resume(actor)
  Actor.registry.shutdownAll
}

If we execute the code, high-priority messages are served first even when low-priority fired before the high-priority messages. This is how the output will appear if executed by sbt on command line:

Work-stealing
‘ExecutorBasedEventDrivenWorkStealingDispatcher’ is one of my favorite dispatcher. It redistributes work to actors that use the same dispatcher and that do not currently have any messages the mailbox. It is a great way to increase performance of the system.

Usual way to use it is to create an Actor companion object to hold the dispatcher and then set it in Actor explicitly.

Lets look at the code example which uses this dispatcher.

package com.meetu.akka.dispatcher

import akka.actor.Actor
import akka.actor.Actor._
import akka.dispatch.Dispatchers

object WorkStealingDispatcherExample extends App {
  val actor = actorOf[SimpleActor]
  actor.start
  actor ! "Hello"
  Actor.registry.shutdownAll
}

object SimpleActor {
  val dispatcher = Dispatchers.newExecutorBasedEventDrivenWorkStealingDispatcher("executordispatcher").build
}

class SimpleActor extends Actor {
  self.dispatcher = SimpleActor.dispatcher
  def receive = {
    case msg => println(msg)
  }
}

Now lets have a look at a code listing for ExecutorDispatcherApplicationExample. It’s constituents are a Scala Object ExecutorDispatcherApplicationExample. Two Pairs of Scala Class and Object (Object is there to share dispatchers) and last three case classes which corresponds to message passing between Actors. Functionally ProcessorUsingExecutorBasedDispatcher is a master actor as it spawns workers and ProcessorUsingExecutorBasedDispatcherWorker is a worker.

package com.meetu.akka.dispatcher

import akka.actor.Actor.actorOf
import akka.actor.Actor
import akka.dispatch.Dispatchers
import akka.routing.CyclicIterator
import akka.routing.Routing

object ExecutorDispatcherApplicationExample extends App {
  val processorUsingExecutorBasedDispatcher = actorOf[ProcessorUsingExecutorBasedDispatcher]
  processorUsingExecutorBasedDispatcher.start
  processorUsingExecutorBasedDispatcher ! new SimpleMessage("Hello Actor!!")
}

object ProcessorUsingExecutorBasedDispatcher {
  val dispatcher = Dispatchers.newExecutorBasedEventDrivenDispatcher("ed1").setCorePoolSize(10).setMaxPoolSize(10).build
}

class ProcessorUsingExecutorBasedDispatcher extends Actor {
  self.dispatcher = ProcessorUsingExecutorBasedDispatcher.dispatcher
  var startTime: Long = -1
  var parallelCounter: Int = 1000

  def receive = {
    case SimpleReply => {
      parallelCounter -= 1
      if (parallelCounter == 0) {
        val endTime = System.currentTimeMillis
        println("Total Time:: " + (endTime - startTime))
        Actor.registry.shutdownAll
      }
    }

    case SimpleMessage(msg) =>
      startTime = System.currentTimeMillis
      val workers = Vector.fill(15)(actorOf[ProcessorUsingExecutorBasedDispatcherWorker].start)
      val router = Routing.loadBalancerActor(CyclicIterator(workers)).start
      for (i <- 0 until parallelCounter) {
        router ! new SimpleRequest(msg)
      }
  }
}

object ProcessorUsingExecutorBasedDispatcherWorker {
  val dispatcher = Dispatchers.newExecutorBasedEventDrivenDispatcher("ed2").setCorePoolSize(10).setMaxPoolSize(10).build
}

class ProcessorUsingExecutorBasedDispatcherWorker extends Actor {
  self.dispatcher = ProcessorUsingExecutorBasedDispatcherWorker.dispatcher
  def receive = {
    case SimpleRequest(msg) => {
      val sleepTime = java.lang.Math.random() * 510
      Thread.sleep(sleepTime.toLong)
      self reply SimpleReply
    }
  }
}

The Application ExecutorDispatcherApplicationExample starts the ProcessorUsingExecutorBasedDispatcher Actor and passes a SimpleMessage to it.

ProcessorUsingExecutorBasedDispatcher Actor uses ExecutorBasedDispatcher and it does so by using a companion object. Please notice that the dispatcher of the object is assigned to self.

...
self.dispatcher = ProcessorUsingExecutorBasedDispatcher.dispatcher
...

The receiving block in ProcessorUsingExecutorBasedDispatcher “case SimpleMessage(msg) =>” starts the workers and a Router over it. It also finally fires “parallelCounter” number of messages to the router.

The Worker Actor ProcessorUsingExecutorBasedDispatcherWorker also uses ExecutorBasedDispatcher. Its received block “case SimpleRequest(msg) =>” replies back to ProcessorUsingExecutorBasedDispatcher after sleeping off for a while.

The Actor ProcessorUsingExecutorBasedDispatcher receives the response and it continue to receive it till all responses are received from Worker Actor. Finally it calculates total time elapsed for parallel process execution.

We also can code this using Work Stealing Dispatcher. The complete code example is given below:

package com.meetu.akka.dispatcher

import akka.actor.Actor.actorOf
import akka.actor.Actor
import akka.dispatch.Dispatchers
import akka.routing.CyclicIterator
import akka.routing.Routing

object WorkStealingDispatcherApplicationExample extends App {
  val processorUsingWorkStealingDispatcher = actorOf[ProcessorUsingWorkStealingDispatcher]
  processorUsingWorkStealingDispatcher.start
  processorUsingWorkStealingDispatcher ! new SimpleMessage("Hello Actor!!")
}

object ProcessorUsingWorkStealingDispatcher {
  val dispatcher = Dispatchers.newExecutorBasedEventDrivenWorkStealingDispatcher("wsd1").setCorePoolSize(10).setMaxPoolSize(10).build
}

class ProcessorUsingWorkStealingDispatcher extends Actor {
  self.dispatcher = ProcessorUsingWorkStealingDispatcher.dispatcher
  var startTime: Long = -1
  var parallelCounter: Int = 1000

  def receive = {
    case SimpleReply => {
      parallelCounter -= 1
      if (parallelCounter == 0) {
        val endTime = System.currentTimeMillis
        println("Total Time:: " + (endTime - startTime))
        Actor.registry.shutdownAll
      }
    }

    case SimpleMessage(msg) =>
      startTime = System.currentTimeMillis
      val workers = Vector.fill(15)(actorOf[ProcessorUsingWorkStealingDispatcherWorker].start)
      val router = Routing.loadBalancerActor(CyclicIterator(workers)).start
      for (i <- 0 until parallelCounter) {
        router ! new SimpleRequest(msg)
      }
  }
}

object ProcessorUsingWorkStealingDispatcherWorker {
  val dispatcher = Dispatchers.newExecutorBasedEventDrivenWorkStealingDispatcher("wsd2").setCorePoolSize(10).setMaxPoolSize(10).build
}

class ProcessorUsingWorkStealingDispatcherWorker extends Actor {
  self.dispatcher = ProcessorUsingWorkStealingDispatcherWorker.dispatcher
  def receive = {
    case SimpleRequest(msg) => {
      val sleepTime = java.lang.Math.random() * 510
      Thread.sleep(sleepTime.toLong)
      self reply SimpleReply
    }
  }
}

case class SimpleRequest(msg: String)

case class SimpleMessage(msg: String)

case class SimpleReply

That is all about dispatchers, they are a powerful way of managing performance, throughput and scalability of the system. All code examples are here at my github repository. It is a sbt project. The command