Routers

Dependency

To use Akka Actor Typed, you must add the following dependency in your project:

sbt

libraryDependencies += "com.typesafe.akka" %% "akka-actor-typed" % "2.5.22"

Maven

<dependency>
  <groupId>com.typesafe.akka</groupId>
  <artifactId>akka-actor-typed_2.12</artifactId>
  <version>2.5.22</version>
</dependency>

Gradle

dependencies {
  compile group: 'com.typesafe.akka', name: 'akka-actor-typed_2.12', version: '2.5.22'
}

Introduction

In some cases it is useful to distribute messages of the same type over a set of actors, so that messages can be processed in parallel - a single actor will only process one message at a time.

The router itself is a behavior that is spawned into a running actor that will then forward any message sent to it to one final recipient out of the set of routees.

There are two kinds of routers included in Akka Typed - the pool router and the group router.

Pool Router

The pool router is created with a routee Behavior and spawns a number of children with that behavior which it will then forward messages to.

If a child is stopped the pool router removes it from its set of routees. When the last child stops the router itself stops. To make a resilient router that deals with failures the routee Behavior must be supervised.

Scala

import akka.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
import akka.actor.typed.Behavior
import akka.actor.typed.SupervisorStrategy
import akka.actor.typed.receptionist.Receptionist
import akka.actor.typed.receptionist.ServiceKey
import akka.actor.typed.scaladsl.Behaviors
import akka.actor.typed.scaladsl.Routers
import org.scalatest.WordSpecLike

object Worker {
  sealed trait Command
  case class DoLog(text: String) extends Command

  val behavior: Behavior[Command] = Behaviors.setup { ctx =>
    ctx.log.info("Starting worker")

    Behaviors.receiveMessage {
      case DoLog(text) =>
        ctx.log.info("Got message {}", text)
        Behaviors.same
    }
  }
}

      // make sure the workers are restarted if they fail
      val supervisedWorker = Behaviors.supervise(Worker.behavior).onFailure[Exception](SupervisorStrategy.restart)
      val pool = Routers.pool(poolSize = 4)(supervisedWorker)
      val router = ctx.spawn(pool, "worker-pool")

      (0 to 10).foreach { n =>
        router ! Worker.DoLog(s"msg $n")
      }

Java

import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.SupervisorStrategy;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.GroupRouter;
import akka.actor.typed.javadsl.PoolRouter;
import akka.actor.typed.javadsl.Routers;
import akka.actor.typed.receptionist.Receptionist;
import akka.actor.typed.receptionist.ServiceKey;

class Worker {
  interface Command {}

  static class DoLog implements Command {
    public final String text;

    public DoLog(String text) {
      this.text = text;
    }
  }

  static final Behavior<Command> behavior =
      Behaviors.setup(
          context -> {
            context.getLog().info("Starting worker");

            return Behaviors.receive(Command.class)
                .onMessage(
                    DoLog.class,
                    (notUsed, doLog) -> {
                      context.getLog().info("Got message {}", doLog.text);
                      return Behaviors.same();
                    })
                .build();
          });
}

        // make sure the workers are restarted if they fail
        Behavior<Worker.Command> supervisedWorker =
            Behaviors.supervise(Worker.behavior).onFailure(SupervisorStrategy.restart());
        PoolRouter<Worker.Command> pool = Routers.pool(4, supervisedWorker);
        ActorRef<Worker.Command> router = context.spawn(pool, "worker-pool");

        for (int i = 0; i < 10; i++) {
          router.tell(new Worker.DoLog("msg " + i));
        }

Group Router

The group router is created with a ServiceKey and uses the receptionist (see Receptionist) to discover available actors for that key and routes messages to one of the currently known registered actors for a key.

Since the receptionist is used this means the group router is cluster aware out of the box and will pick up routees registered on any node in the cluster (there is currently no logic to avoid routing to unreachable nodes, see #26355).

It also means that the set of routees is eventually consistent, and that immediately when the group router is started the set of routees it knows about is empty. When the set of routees is empty messages sent to the router is forwarded to dead letters.

Scala

// this would likely happen elsewhere - if we create it locally we
// can just as well use a pool
val worker = ctx.spawn(Worker.behavior, "worker")
ctx.system.receptionist ! Receptionist.Register(serviceKey, worker)

val group = Routers.group(serviceKey);
val router = ctx.spawn(group, "worker-group");

// note that since registration of workers goes through the receptionist there is no
// guarantee the router has seen any workers yet if we hit it directly like this and
// these messages may end up in dead letters - in a real application you would not use
// a group router like this - it is to keep the sample simple
(0 to 10).foreach { n =>
  router ! Worker.DoLog(s"msg $n")
}

Java

// this would likely happen elsewhere - if we create it locally we
// can just as well use a pool
ActorRef<Worker.Command> worker = context.spawn(Worker.behavior, "worker");
context.getSystem().receptionist().tell(Receptionist.register(serviceKey, worker));

GroupRouter<Worker.Command> group = Routers.group(serviceKey);
ActorRef<Worker.Command> router = context.spawn(group, "worker-group");

// note that since registration of workers goes through the receptionist there is no
// guarantee the router has seen any workers yet if we hit it directly like this and
// these messages may end up in dead letters - in a real application you would not use
// a group router like this - it is to keep the sample simple
for (int i = 0; i < 10; i++) {
  router.tell(new Worker.DoLog("msg " + i));
}

Routing strategies

There are two different strategies for selecting what routee a message is forwarded to that can be selected from the router before spawning it:

Scala

val alternativePool = pool.withPoolSize(2).withRoundRobinRouting()

Java

PoolRouter<Worker.Command> alternativePool = pool.withPoolSize(2).withRoundRobinRouting();

Round Robin

Rotates over the set of routees making sure that if there are n routees, then for n messages sent through the router, each actor is forwarded one message.

This is the default for pool routers.

Random

Randomly selects a routee when a message is sent through the router.

This is the default for group routers.

Routers and performance

Note that if the routees are sharing a resource, the resource will determine if increasing the number of actors will actually give higher throughput or faster answers. For example if the routees are CPU bound actors it will not give better performance to create more routees than there are threads to execute the actors.

Since the router itself is an actor and has a mailbox this means that messages are routed sequentially to the routees where it can be processed in parallel (depending on the available threads in the dispatcher). In a high throughput use cases the sequential routing could be a bottle neck. Akka Typed does not provide an optimized tool for this.