Classic Dispatchers

For the full documentation of this feature and for new projects see Dispatchers.

Dependency

Dispatchers are part of core Pekko, which means that they are part of the pekko-actor dependency:

sbt

val PekkoVersion = "1.1.2+29-e21fa9eb-SNAPSHOT"
libraryDependencies += "org.apache.pekko" %% "pekko-actor" % PekkoVersion

Maven

<properties>
  <scala.binary.version>2.13</scala.binary.version>
</properties>
<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>org.apache.pekko</groupId>
      <artifactId>pekko-bom_${scala.binary.version}</artifactId>
      <version>1.1.2+29-e21fa9eb-SNAPSHOT</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>
<dependencies>
  <dependency>
    <groupId>org.apache.pekko</groupId>
    <artifactId>pekko-actor_${scala.binary.version}</artifactId>
  </dependency>
</dependencies>

Gradle

def versions = [
  ScalaBinary: "2.13"
]
dependencies {
  implementation platform("org.apache.pekko:pekko-bom_${versions.ScalaBinary}:1.1.2+29-e21fa9eb-SNAPSHOT")

  implementation "org.apache.pekko:pekko-actor_${versions.ScalaBinary}"
}

Looking up a Dispatcher

Dispatchers implement the ExecutionContextExecutor interface and can thus be used to run FutureCompletableFuture invocations etc.

Scala

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

Java

copysource// this is scala.concurrent.ExecutionContextExecutor, which implements
// both scala.concurrent.ExecutionContext (for use with Futures, Scheduler, etc.)
// and java.util.concurrent.Executor (for use with CompletableFuture etc.)
final ExecutionContextExecutor ex = system.dispatchers().lookup("my-dispatcher");

Setting the dispatcher for an Actor

So in case you want to give your ActorActor a different dispatcher than the default, you need to do two things, of which the first is to configure the dispatcher:

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

Another example that uses the “thread-pool-executor”:

copysourceblocking-io-dispatcher {
  type = Dispatcher
  executor = "thread-pool-executor"
  thread-pool-executor {
    fixed-pool-size = 32
  }
  throughput = 1
}

For more options, see Dispatchers and the default-dispatcher section of the configuration.

Then you create the actor as usual and define the dispatcher in the deployment configuration.

Scala

copysourceimport org.apache.pekko.actor.Props
val myActor = context.actorOf(Props[MyActor](), "myactor")

Java

copysourceActorRef myActor = system.actorOf(Props.create(MyActor.class), "myactor");

copysourcepekko.actor.deployment {
  /myactor {
    dispatcher = my-dispatcher
  }
}

An alternative to the deployment configuration is to define the dispatcher in code. If you define the dispatcher in the deployment configuration then this value will be used instead of programmatically provided parameter.

Scala

copysourceimport org.apache.pekko.actor.Props
val myActor =
  context.actorOf(Props[MyActor]().withDispatcher("my-dispatcher"), "myactor1")

Java

copysourceActorRef myActor =
    system.actorOf(Props.create(MyActor.class).withDispatcher("my-dispatcher"), "myactor3");