Routing

Pool

The following code and configuration snippets show how to create a round-robin router that forwards messages to five Worker routees. The routees will be created as the router's children.

akka.actor.deployment {
  /parent/router1 {
    router = round-robin-pool
    nr-of-instances = 5
  }
}

ActorRef router1 =
  getContext().actorOf(FromConfig.getInstance().props(Props.create(Worker.class)), 
    "router1");

Here is the same example, but with the router configuration provided programmatically instead of from configuration.

ActorRef router2 =
  getContext().actorOf(new RoundRobinPool(5).props(Props.create(Worker.class)), 
    "router2");

Address[] addresses = {
  new Address("akka.tcp", "remotesys", "otherhost", 1234),
  AddressFromURIString.parse("akka.tcp://othersys@anotherhost:1234")};
ActorRef routerRemote = system.actorOf(
  new RemoteRouterConfig(new RoundRobinPool(5), addresses).props(
    Props.create(Echo.class)));

getSender().tell("reply", getSelf());

getSender().tell("reply", getContext().parent());

final SupervisorStrategy strategy =
  new OneForOneStrategy(5, Duration.create(1, TimeUnit.MINUTES),
    Collections.<Class<? extends Throwable>>singletonList(Exception.class));
final ActorRef router = system.actorOf(new RoundRobinPool(5).
  withSupervisorStrategy(strategy).props(Props.create(Echo.class)));

Group

Sometimes, rather than having the router actor create its routees, it is desirable to create routees separately and provide them to the router for its use. You can do this by passing an paths of the routees to the router's configuration. Messages will be sent with ActorSelection to these paths.

The example below shows how to create a router by providing it with the path strings of three routee actors.

akka.actor.deployment {
  /parent/router3 {
    router = round-robin-group
    routees.paths = ["/user/workers/w1", "/user/workers/w2", "/user/workers/w3"]
  }
}

ActorRef router3 =
  getContext().actorOf(FromConfig.getInstance().props(), "router3");

Here is the same example, but with the router configuration provided programmatically instead of from configuration.

ActorRef router4 =
  getContext().actorOf(new RoundRobinGroup(paths).props(), "router4");

The routee actors are created externally from the router:

system.actorOf(Props.create(Workers.class), "workers");

public class Workers extends UntypedActor {
  @Override public void preStart() {
    getContext().actorOf(Props.create(Worker.class), "w1");
    getContext().actorOf(Props.create(Worker.class), "w2");
    getContext().actorOf(Props.create(Worker.class), "w3");
  }
  // ...

The paths may contain protocol and address information for actors running on remote hosts. Remoting requires the akka-remote module to be included in the classpath.

akka.actor.deployment {
  /parent/remoteGroup {
    router = round-robin-group
    routees.paths = [
      "akka.tcp://app@10.0.0.1:2552/user/workers/w1",
      "akka.tcp://app@10.0.0.2:2552/user/workers/w1",
      "akka.tcp://app@10.0.0.3:2552/user/workers/w1"]
  }
}

RoundRobinPool and RoundRobinGroup

Routes in a round-robin fashion to its routees.

RoundRobinPool defined in configuration:

akka.actor.deployment {
  /parent/router1 {
    router = round-robin-pool
    nr-of-instances = 5
  }
}

ActorRef router1 =
  getContext().actorOf(FromConfig.getInstance().props(Props.create(Worker.class)), 
    "router1");

RoundRobinPool defined in code:

ActorRef router2 =
  getContext().actorOf(new RoundRobinPool(5).props(Props.create(Worker.class)), 
    "router2");

RoundRobinGroup defined in configuration:

akka.actor.deployment {
  /parent/router3 {
    router = round-robin-group
    routees.paths = ["/user/workers/w1", "/user/workers/w2", "/user/workers/w3"]
  }
}

ActorRef router3 =
  getContext().actorOf(FromConfig.getInstance().props(), "router3");

RoundRobinGroup defined in code:

List<String> paths = Arrays.asList("/user/workers/w1", "/user/workers/w2", 
  "/user/workers/w3");
ActorRef router4 =
  getContext().actorOf(new RoundRobinGroup(paths).props(), "router4");

RandomPool and RandomGroup

This router type selects one of its routees randomly for each message.

RandomPool defined in configuration:

akka.actor.deployment {
  /parent/router5 {
    router = random-pool
    nr-of-instances = 5
  }
}

ActorRef router5 =
  getContext().actorOf(FromConfig.getInstance().props(
    Props.create(Worker.class)), "router5");

RandomPool defined in code:

ActorRef router6 =
  getContext().actorOf(new RandomPool(5).props(Props.create(Worker.class)), 
    "router6");

RandomGroup defined in configuration:

akka.actor.deployment {
  /parent/router7 {
    router = random-group
    routees.paths = ["/user/workers/w1", "/user/workers/w2", "/user/workers/w3"]
  }
}

ActorRef router7 =
  getContext().actorOf(FromConfig.getInstance().props(), "router7");

RandomGroup defined in code:

List<String> paths = Arrays.asList("/user/workers/w1", "/user/workers/w2", 
  "/user/workers/w3");
ActorRef router8 =
  getContext().actorOf(new RandomGroup(paths).props(), "router8");

BalancingPool

A Router that will try to redistribute work from busy routees to idle routees. All routees share the same mailbox.

BalancingPool defined in configuration:

akka.actor.deployment {
  /parent/router9 {
    router = balancing-pool
    nr-of-instances = 5
  }
}

ActorRef router9 =
  getContext().actorOf(FromConfig.getInstance().props(
    Props.create(Worker.class)), "router9");

BalancingPool defined in code:

ActorRef router10 =
  getContext().actorOf(new BalancingPool(5).props(
    Props.create(Worker.class)), "router10");

Addition configuration for the balancing dispatcher, which is used by the pool, can be configured in the pool-dispatcher section of the router deployment configuration.

akka.actor.deployment {
  /parent/router9b {
    router = balancing-pool
    nr-of-instances = 5
    pool-dispatcher {
      attempt-teamwork = off
    }
  }
}

The BalancingPool automatically uses a special BalancingDispatcher for its routees - disregarding any dispatcher that is set on the routee Props object. This is needed in order to implement the balancing semantics via sharing the same mailbox by all the routees.

While it is not possible to change the dispatcher used by the routees, it is possible to fine tune the used executor. By default the fork-join-dispatcher is used and can be configured as explained in Dispatchers. In situations where the routees are expected to perform blocking operations it may be useful to replace it with a thread-pool-executor hinting the number of allocated threads explicitly:

akka.actor.deployment {
  /parent/router10b {
    router = balancing-pool
    nr-of-instances = 5
    pool-dispatcher {
      executor = "thread-pool-executor"

      # allocate exactly 5 threads for this pool
      thread-pool-executor {
        core-pool-size-min = 5
        core-pool-size-max = 5
      }
    }
  }
}

It is also possible to change the mailbox used by the balancing dispatcher for scenarios where the default unbounded mailbox is not well suited. An example of such a scenario could arise whether there exists the need to manage priority for each message. You can then implement a priority mailbox and configure your dispatcher:

akka.actor.deployment {
  /parent/router10c {
    router = balancing-pool
    nr-of-instances = 5
    pool-dispatcher {
      mailbox = myapp.myprioritymailbox
    }
  }
}

There is no Group variant of the BalancingPool.

SmallestMailboxPool

A Router that tries to send to the non-suspended child routee with fewest messages in mailbox. The selection is done in this order:

• pick any idle routee (not processing message) with empty mailbox
• pick any routee with empty mailbox
• pick routee with fewest pending messages in mailbox
• pick any remote routee, remote actors are consider lowest priority, since their mailbox size is unknown

SmallestMailboxPool defined in configuration:

akka.actor.deployment {
  /parent/router11 {
    router = smallest-mailbox-pool
    nr-of-instances = 5
  }
}

ActorRef router11 =
  getContext().actorOf(FromConfig.getInstance().props(
    Props.create(Worker.class)), "router11");

SmallestMailboxPool defined in code:

ActorRef router12 =
  getContext().actorOf(new SmallestMailboxPool(5).props(
    Props.create(Worker.class)), "router12");

There is no Group variant of the SmallestMailboxPool because the size of the mailbox and the internal dispatching state of the actor is not practically available from the paths of the routees.

BroadcastPool and BroadcastGroup

A broadcast router forwards the message it receives to all its routees.

BroadcastPool defined in configuration:

akka.actor.deployment {
  /parent/router13 {
    router = broadcast-pool
    nr-of-instances = 5
  }
}

ActorRef router13 =
  getContext().actorOf(FromConfig.getInstance().props(
    Props.create(Worker.class)), "router13");

BroadcastPool defined in code:

ActorRef router14 =
  getContext().actorOf(new BroadcastPool(5).props(Props.create(Worker.class)),
    "router14");

BroadcastGroup defined in configuration:

akka.actor.deployment {
  /parent/router15 {
    router = broadcast-group
    routees.paths = ["/user/workers/w1", "/user/workers/w2", "/user/workers/w3"]
  }
}

ActorRef router15 =
  getContext().actorOf(FromConfig.getInstance().props(), "router15");

BroadcastGroup defined in code:

List<String> paths = Arrays.asList("/user/workers/w1", "/user/workers/w2", 
  "/user/workers/w3");
ActorRef router16 =
  getContext().actorOf(new BroadcastGroup(paths).props(), "router16");

ScatterGatherFirstCompletedPool and ScatterGatherFirstCompletedGroup

The ScatterGatherFirstCompletedRouter will send the message on to all its routees. It then waits for first reply it gets back. This result will be sent back to original sender. Other replies are discarded.

It is expecting at least one reply within a configured duration, otherwise it will reply with akka.pattern.AskTimeoutException in a akka.actor.Status.Failure.

ScatterGatherFirstCompletedPool defined in configuration:

akka.actor.deployment {
  /parent/router17 {
    router = scatter-gather-pool
    nr-of-instances = 5
    within = 10 seconds
  }
}

ActorRef router17 =
  getContext().actorOf(FromConfig.getInstance().props(
    Props.create(Worker.class)), "router17");

ScatterGatherFirstCompletedPool defined in code:

FiniteDuration within = FiniteDuration.create(10, TimeUnit.SECONDS); 
ActorRef router18 =
  getContext().actorOf(new ScatterGatherFirstCompletedPool(5, within).props(
    Props.create(Worker.class)), "router18");

ScatterGatherFirstCompletedGroup defined in configuration:

akka.actor.deployment {
  /parent/router19 {
    router = scatter-gather-group
    routees.paths = ["/user/workers/w1", "/user/workers/w2", "/user/workers/w3"]
    within = 10 seconds
  }
}

ActorRef router19 =
  getContext().actorOf(FromConfig.getInstance().props(), "router19");

ScatterGatherFirstCompletedGroup defined in code:

List<String> paths = Arrays.asList("/user/workers/w1", "/user/workers/w2", 
  "/user/workers/w3");
FiniteDuration within2 = FiniteDuration.create(10, TimeUnit.SECONDS);
ActorRef router20 =
  getContext().actorOf(new ScatterGatherFirstCompletedGroup(paths, within2).props(), 
    "router20");

TailChoppingPool and TailChoppingGroup

The TailChoppingRouter will first send the message to one, randomly picked, routee and then after a small delay to a second routee (picked randomly from the remaining routees) and so on. It waits for first reply it gets back and forwards it back to original sender. Other replies are discarded.

The goal of this router is to decrease latency by performing redundant queries to multiple routees, assuming that one of the other actors may still be faster to respond than the initial one.

This optimisation was described nicely in a blog post by Peter Bailis: Doing redundant work to speed up distributed queries.

TailChoppingPool defined in configuration:

akka.actor.deployment {
  /parent/router21 {
    router = tail-chopping-pool
    nr-of-instances = 5
    within = 10 seconds
    tail-chopping-router.interval = 20 milliseconds
  }
}

ActorRef router21 =
  getContext().actorOf(FromConfig.getInstance().props(
    Props.create(Worker.class)), "router21");

TailChoppingPool defined in code:

FiniteDuration within3 = FiniteDuration.create(10, TimeUnit.SECONDS);
FiniteDuration interval = FiniteDuration.create(20, TimeUnit.MILLISECONDS);
ActorRef router22 =
  getContext().actorOf(new TailChoppingPool(5, within3, interval).props(
    Props.create(Worker.class)), "router22");

TailChoppingGroup defined in configuration:

akka.actor.deployment {
  /parent/router23 {
    router = tail-chopping-group
    routees.paths = ["/user/workers/w1", "/user/workers/w2", "/user/workers/w3"]
    within = 10 seconds
    tail-chopping-router.interval = 20 milliseconds
  }
}

ActorRef router23 =
  getContext().actorOf(FromConfig.getInstance().props(), "router23");

TailChoppingGroup defined in code:

List<String> paths = Arrays.asList("/user/workers/w1", "/user/workers/w2", 
  "/user/workers/w3");
FiniteDuration within4 = FiniteDuration.create(10, TimeUnit.SECONDS);
FiniteDuration interval2 = FiniteDuration.create(20, TimeUnit.MILLISECONDS);
ActorRef router24 =
  getContext().actorOf(new TailChoppingGroup(paths, within4, interval2).props(),
    "router24");

ConsistentHashingPool and ConsistentHashingGroup

The ConsistentHashingPool uses consistent hashing to select a routee based on the sent message. This article gives good insight into how consistent hashing is implemented.

There is 3 ways to define what data to use for the consistent hash key.

• You can define withHashMapper of the router to map incoming messages to their consistent hash key. This makes the decision transparent for the sender.
• The messages may implement akka.routing.ConsistentHashingRouter.ConsistentHashable. The key is part of the message and it's convenient to define it together with the message definition.
• The messages can be wrapped in a akka.routing.ConsistentHashingRouter.ConsistentHashableEnvelope to define what data to use for the consistent hash key. The sender knows the key to use.

These ways to define the consistent hash key can be use together and at the same time for one router. The withHashMapper is tried first.

Code example:

public class Cache extends UntypedActor {
  Map<String, String> cache = new HashMap<String, String>();

  public void onReceive(Object msg) {
    if (msg instanceof Entry) {
      Entry entry = (Entry) msg;
      cache.put(entry.key, entry.value);
    } else if (msg instanceof Get) {
      Get get = (Get) msg;
      Object value = cache.get(get.key);
      getSender().tell(value == null ? NOT_FOUND : value, 
        getContext().self());
    } else if (msg instanceof Evict) {
      Evict evict = (Evict) msg;
      cache.remove(evict.key);
    } else {
      unhandled(msg);
    }
  }
}

public final class Evict implements Serializable {
  private static final long serialVersionUID = 1L;
  public final String key;
  public Evict(String key) {
    this.key = key;
  }
}

public final class Get implements Serializable, ConsistentHashable {
  private static final long serialVersionUID = 1L;
  public final String key;
  public Get(String key) {
    this.key = key;
  }
  public Object consistentHashKey() {
    return key;
  } 
}

public final class Entry implements Serializable {
  private static final long serialVersionUID = 1L;
  public final String key;
  public final String value;
  public Entry(String key, String value) {
    this.key = key;
    this.value = value;
  }
}

public final String NOT_FOUND = "NOT_FOUND";

final ConsistentHashMapper hashMapper = new ConsistentHashMapper() {
  @Override
  public Object hashKey(Object message) {
    if (message instanceof Evict) {
      return ((Evict) message).key;
    } else {
      return null;
    }
  }
};

ActorRef cache = system.actorOf(
    new ConsistentHashingPool(10).withHashMapper(hashMapper).props(
        Props.create(Cache.class)), 
  "cache");

cache.tell(new ConsistentHashableEnvelope(
  new Entry("hello", "HELLO"), "hello"), getRef());
cache.tell(new ConsistentHashableEnvelope(
  new Entry("hi", "HI"), "hi"), getRef());

cache.tell(new Get("hello"), getRef());
expectMsgEquals("HELLO");

cache.tell(new Get("hi"), getRef());
expectMsgEquals("HI");

cache.tell(new Evict("hi"), getRef());
cache.tell(new Get("hi"), getRef());
expectMsgEquals(NOT_FOUND);

In the above example you see that the Get message implements ConsistentHashable itself, while the Entry message is wrapped in a ConsistentHashableEnvelope. The Evict message is handled by the hashMapping partial function.

ConsistentHashingPool defined in configuration:

akka.actor.deployment {
  /parent/router25 {
    router = consistent-hashing-pool
    nr-of-instances = 5
    virtual-nodes-factor = 10
  }
}

ActorRef router25 =
  getContext().actorOf(FromConfig.getInstance().props(Props.create(Worker.class)),
    "router25");

ConsistentHashingPool defined in code:

ActorRef router26 =
  getContext().actorOf(new ConsistentHashingPool(5).props(
    Props.create(Worker.class)), "router26");

ConsistentHashingGroup defined in configuration:

akka.actor.deployment {
  /parent/router27 {
    router = consistent-hashing-group
    routees.paths = ["/user/workers/w1", "/user/workers/w2", "/user/workers/w3"]
    virtual-nodes-factor = 10
  }
}

ActorRef router27 =
  getContext().actorOf(FromConfig.getInstance().props(), "router27");

ConsistentHashingGroup defined in code:

List<String> paths = Arrays.asList("/user/workers/w1", "/user/workers/w2", 
  "/user/workers/w3");
ActorRef router28 =
  getContext().actorOf(new ConsistentHashingGroup(paths).props(), "router28");

virtual-nodes-factor is the number of virtual nodes per routee that is used in the consistent hash node ring to make the distribution more uniform.

Broadcast Messages

A Broadcast message can be used to send a message to all of a router's routees. When a router receives a Broadcast message, it will broadcast that message's payload to all routees, no matter how that router would normally route its messages.

The example below shows how you would use a Broadcast message to send a very important message to every routee of a router.

router.tell(new Broadcast("Watch out for Davy Jones' locker"), getTestActor());

In this example the router receives the Broadcast message, extracts its payload ("Watch out for Davy Jones' locker"), and then sends the payload on to all of the router's routees. It is up to each routee actor to handle the received payload message.

PoisonPill Messages

A PoisonPill message has special handling for all actors, including for routers. When any actor receives a PoisonPill message, that actor will be stopped. See the PoisonPill documentation for details.

router.tell(PoisonPill.getInstance(), getTestActor());

For a router, which normally passes on messages to routees, it is important to realise that PoisonPill messages are processed by the router only. PoisonPill messages sent to a router will not be sent on to routees.

However, a PoisonPill message sent to a router may still affect its routees, because it will stop the router and when the router stops it also stops its children. Stopping children is normal actor behavior. The router will stop routees that it has created as children. Each child will process its current message and then stop. This may lead to some messages being unprocessed. See the documentation on Stopping actors for more information.

If you wish to stop a router and its routees, but you would like the routees to first process all the messages currently in their mailboxes, then you should not send a PoisonPill message to the router. Instead you should wrap a PoisonPill message inside a Broadcast message so that each routee will receive the PoisonPill message. Note that this will stop all routees, even if the routees aren't children of the router, i.e. even routees programmatically provided to the router.

router.tell(new Broadcast(PoisonPill.getInstance()), getTestActor());

With the code shown above, each routee will receive a PoisonPill message. Each routee will continue to process its messages as normal, eventually processing the PoisonPill. This will cause the routee to stop. After all routees have stopped the router will itself be stopped automatically unless it is a dynamic router, e.g. using a resizer.

Kill Messages

Kill messages are another type of message that has special handling. See Killing an Actor for general information about how actors handle Kill messages.

When a Kill message is sent to a router the router processes the message internally, and does not send it on to its routees. The router will throw an ActorKilledException and fail. It will then be either resumed, restarted or terminated, depending how it is supervised.

Routees that are children of the router will also be suspended, and will be affected by the supervision directive that is applied to the router. Routees that are not the routers children, i.e. those that were created externally to the router, will not be affected.

router.tell(Kill.getInstance(), getTestActor());

As with the PoisonPill message, there is a distinction between killing a router, which indirectly kills its children (who happen to be routees), and killing routees directly (some of whom may not be children.) To kill routees directly the router should be sent a Kill message wrapped in a Broadcast message.

router.tell(new Broadcast(Kill.getInstance()), getTestActor());

Management Messages

• Sending akka.routing.GetRoutees to a router actor will make it send back its currently used routees in a akka.routing.Routees message.
• Sending akka.routing.AddRoutee to a router actor will add that routee to its collection of routees.
• Sending akka.routing.RemoveRoutee to a router actor will remove that routee to its collection of routees.
• Sending akka.routing.AdjustPoolSize to a pool router actor will add or remove that number of routees to its collection of routees.

These management messages may be handled after other messages, so if you send AddRoutee immediately followed by an ordinary message you are not guaranteed that the routees have been changed when the ordinary message is routed. If you need to know when the change has been applied you can send AddRoutee followed by GetRoutees and when you receive the Routees reply you know that the preceding change has been applied.

Default Resizer

The default resizer ramps up and down pool size based on pressure, measured by the percentage of busy routees in the pool. It ramps up pool size if the pressure is higher than a certain threshold and backs off if the pressure is lower than certain threshold. Both thresholds are configurable.

Pool with default resizer defined in configuration:

akka.actor.deployment {
  /parent/router29 {
    router = round-robin-pool
    resizer {
      lower-bound = 2
      upper-bound = 15
      messages-per-resize = 100
    }
  }
}

ActorRef router29 =
  getContext().actorOf(FromConfig.getInstance().props(
    Props.create(Worker.class)), "router29");

Several more configuration options are available and described in akka.actor.deployment.default.resizer section of the reference 環境設定.

Pool with resizer defined in code:

DefaultResizer resizer = new DefaultResizer(2, 15);
ActorRef router30 =
  getContext().actorOf(new RoundRobinPool(5).withResizer(resizer).props(
    Props.create(Worker.class)), "router30");

It is also worth pointing out that if you define the ``router`` in the configuration file then this value will be used instead of any programmatically sent parameters.

Optimal Size Exploring Resizer

The OptimalSizeExploringResizer resizes the pool to an optimal size that provides the most message throughput.

It achieves this by keeping track of message throughput at each pool size and performing one of the following three resizing operations periodically:

• Downsize if it hasn't seen all routees ever fully utilized for a period of time.
• Explore to a random nearby pool size to try and collect throughput metrics.
• Optimize to a nearby pool size with a better (than any other nearby sizes) throughput metrics.

When the pool is fully-utilized (i.e. all routees are busy), it randomly choose between exploring and optimizing. When the pool has not been fully-utilized for a period of time, it will downsize the pool to the last seen max utilization multiplied by a configurable ratio.

By constantly exploring and optimizing, the resizer will eventually walk to the optimal size and remain nearby. When the optimal size changes it will start walking towards the new one. This resizer works best when you expect the pool size to performance function to be a convex function. For example, when you have a CPU bound tasks, the optimal size is bound to the number of CPU cores. When your task is IO bound, the optimal size is bound to optimal number of concurrent connections to that IO service - e.g. a 4 node elastic search cluster may handle 4-8 concurrent requests at optimal speed.

It keeps a performance log so it's stateful as well as having a larger memory footprint than the default Resizer. The memory usage is O(n) where n is the number of sizes you allow, i.e. upperBound - lowerBound.

Pool with OptimalSizeExploringResizer defined in configuration:

akka.actor.deployment {
  /parent/router31 {
    router = round-robin-pool
    optimal-size-exploring-resizer {
      enabled = on
      action-interval = 5s
      downsize-after-underutilized-for = 72h
    }
  }
}

ActorRef router31 =
  getContext().actorOf(FromConfig.getInstance().props(
    Props.create(Worker.class)), "router31");

Several more configuration options are available and described in akka.actor.deployment.default.optimal-size-exploring-resizer section of the reference 環境設定.