raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 11

Upgraded routing documentation to Akka 2.0. See #1063

Browse source
This commit is contained in:
Henrik Engstrom 2011-12-15 15:28:21 +01:00
parent 73b79d6e3e
commit 41ce42c8f7
8 changed files with 495 additions and 189 deletions
Download patch file Download diff file Expand all files Collapse all files

88
akka-actor-tests/src/test/scala/akka/routing/RoutingSpec.scala
View file

@ -339,4 +339,92 @@ class RoutingSpec extends AkkaSpec with DefaultTimeout with ImplicitSender {
}
}), "Actor:" + id)
}
"custom router" must {
"be started when constructed" in {
val routedActor = system.actorOf(Props(new TestActor).withRouter(VoteCountRouter()))
routedActor.isTerminated must be(false)
}
"count votes as intended - not as in Florida" in {
val routedActor = system.actorOf(Props(new TestActor).withRouter(VoteCountRouter()))
routedActor ! DemocratVote
routedActor ! DemocratVote
routedActor ! RepublicanVote
routedActor ! DemocratVote
routedActor ! RepublicanVote
val democratsResult = (routedActor ? DemocratCountResult)
val republicansResult = (routedActor ? RepublicanCountResult)
Await.result(democratsResult, 1 seconds)
Await.result(republicansResult, 1 seconds)
democratsResult.value must be(Some(Right(3)))
republicansResult.value must be(Some(Right(2)))
}
// DO NOT CHANGE THE COMMENTS BELOW AS THEY ARE USED IN THE DOCUMENTATION
//#CustomRouter
//#crMessages
case object DemocratVote
case object DemocratCountResult
case object RepublicanVote
case object RepublicanCountResult
//#crMessages
//#crActors
class DemocratActor extends Actor {
val counter = new AtomicInteger(0)
def receive = {
case DemocratVote ⇒ counter.incrementAndGet()
case DemocratCountResult ⇒ sender ! counter.get
}
}
class RepublicanActor extends Actor {
val counter = new AtomicInteger(0)
def receive = {
case RepublicanVote ⇒ counter.incrementAndGet()
case RepublicanCountResult ⇒ sender ! counter.get
}
}
//#crActors
//#crRouter
case class VoteCountRouter(nrOfInstances: Int = 0, targets: Iterable[String] = Nil)
extends RouterConfig {
//#crRoute
def createRoute(props: Props,
actorContext: ActorContext,
ref: RoutedActorRef): Route = {
val democratActor = actorContext.actorOf(Props(new DemocratActor), "d")
val republicanActor = actorContext.actorOf(Props(new RepublicanActor), "r")
val routees = Vector[ActorRef](democratActor, republicanActor)
//#crRegisterRoutees
registerRoutees(actorContext, routees)
//#crRegisterRoutees
//#crRoutingLogic
{
case (sender, message) ⇒
message match {
case DemocratVote | DemocratCountResult ⇒
List(Destination(sender, democratActor))
case RepublicanVote | RepublicanCountResult ⇒
List(Destination(sender, republicanActor))
}
}
//#crRoutingLogic
}
//#crRoute
}
//#crRouter
//#CustomRouter
}
}

21
akka-docs/scala/code/akka/docs/routing/ActorPoolExample.scala Normal file
View file

@ -0,0 +1,21 @@
package akka.docs.routing
import akka.routing.{ BasicNoBackoffFilter, SmallestMailboxSelector, DefaultActorPool }
import akka.actor.{ ActorRef, Props, Actor }
//#testPool
class TestPool extends Actor with DefaultActorPool with SmallestMailboxSelector with BasicNoBackoffFilter {
def capacity(delegates: Seq[ActorRef]) = 5
protected def receive = _route
def rampupRate = 0.1
def selectionCount = 1
def partialFill = true
def instance(defaults: Props) = context.actorOf(defaults.withCreator(new Actor {
def receive = {
case _ ⇒ // do something
}
}))
}
//#testPool

23
akka-docs/scala/code/akka/docs/routing/BoundedCapacitorExample.scala Normal file
View file

@ -0,0 +1,23 @@
package akka.docs.routing
import akka.actor.ActorRef
//#boundedCapacitor
trait BoundedCapacitor {
def lowerBound: Int
def upperBound: Int
def capacity(delegates: Seq[ActorRef]): Int = {
val current = delegates length
var delta = _eval(delegates)
val proposed = current + delta
if (proposed < lowerBound) delta += (lowerBound - proposed)
else if (proposed > upperBound) delta -= (proposed - upperBound)
delta
}
protected def _eval(delegates: Seq[ActorRef]): Int
}
//#boundedCapacitor

16
akka-docs/scala/code/akka/docs/routing/CapacityStrategyExample.scala Normal file
View file

@ -0,0 +1,16 @@
package akka.docs.routing
import akka.routing.ActorPool
import akka.actor.ActorRef
//#capacityStrategy
trait CapacityStrategy {
import ActorPool._
def pressure(delegates: Seq[ActorRef]): Int
def filter(pressure: Int, capacity: Int): Int
protected def _eval(delegates: Seq[ActorRef]): Int =
filter(pressure(delegates), delegates.size)
}
//#capacityStrategy

78
akka-docs/scala/code/akka/docs/routing/RouterTypeExample.scala Normal file
View file

@ -0,0 +1,78 @@
package akka.docs.routing
import akka.routing.{ ScatterGatherFirstCompletedRouter, BroadcastRouter, RandomRouter, RoundRobinRouter }
import annotation.tailrec
import akka.actor.{ Props, Actor }
import akka.util.duration._
import akka.dispatch.Await
case class FibonacciNumber(nbr: Int)
//#printlnActor
class PrintlnActor extends Actor {
def receive = {
case msg ⇒
println("Received message '%s' in actor %s".format(msg, self.path.name))
}
}
//#printlnActor
//#fibonacciActor
class FibonacciActor extends Actor {
def receive = {
case FibonacciNumber(nbr) ⇒ sender.tell(fibonacci(nbr))
}
private def fibonacci(n: Int): Int = {
@tailrec
def fib(n: Int, b: Int, a: Int): Int = n match {
case 0 ⇒ a
case _ ⇒ fib(n - 1, a + b, b)
}
fib(n, 1, 0)
}
}
//#fibonacciActor
//#parentActor
class ParentActor extends Actor {
def receive = {
case "rrr" ⇒
//#roundRobinRouter
val roundRobinRouter =
context.actorOf(Props[PrintlnActor].withRouter(RoundRobinRouter()), "router")
1 to 10 foreach {
i ⇒ roundRobinRouter ! i
}
//#roundRobinRouter
case "rr" ⇒
//#randomRouter
val randomRouter =
context.actorOf(Props[PrintlnActor].withRouter(RandomRouter()), "router")
1 to 10 foreach {
i ⇒ randomRouter ! i
}
//#randomRouter
case "br" ⇒
//#broadcastRouter
val broadcastRouter =
context.actorOf(Props[PrintlnActor].withRouter(BroadcastRouter()), "router")
broadcastRouter ! "this is a broadcast message"
//#broadcastRouter
case "sgfcr" ⇒
//#scatterGatherFirstCompletedRouter
val scatterGatherFirstCompletedRouter = context.actorOf(
Props[FibonacciActor].withRouter(ScatterGatherFirstCompletedRouter()),
"router")
implicit val timeout = context.system.settings.ActorTimeout
val futureResult = scatterGatherFirstCompletedRouter ? FibonacciNumber(10)
val result = Await.result(futureResult, timeout.duration)
//#scatterGatherFirstCompletedRouter
println("The result of calculating Fibonacci for 10 is %d".format(result))
}
}
//#parentActor

22
akka-docs/scala/code/akka/docs/routing/RouterViaConfigExample.scala Normal file
View file

@ -0,0 +1,22 @@
package akka.docs.routing
import com.typesafe.config.{ ConfigFactory, Config }
import akka.actor.{ Actor, Props, ActorSystem }
import akka.routing.RoundRobinRouter
case class Message(nbr: Int)
class ExampleActor extends Actor {
def receive = {
case Message(nbr) ⇒ println("Received %s in router %s".format(nbr, self.path.name))
}
}
object RouterWithConfigExample extends App {
val system = ActorSystem("Example")
//#configurableRouting
val router = system.actorOf(Props[PrintlnActor].withRouter(RoundRobinRouter()),
"exampleActor")
//#configurableRouting
1 to 10 foreach { i ⇒ router ! Message(i) }
}

31
akka-docs/scala/code/akka/docs/routing/RouterViaProgramExample.scala Normal file
View file

@ -0,0 +1,31 @@
package akka.docs.routing
import akka.routing.RoundRobinRouter
import akka.actor.{ ActorRef, Props, Actor, ActorSystem }
case class Message1(nbr: Int)
class ExampleActor1 extends Actor {
def receive = {
case Message1(nbr) ⇒ println("Received %s in router %s".format(nbr, self.path.name))
}
}
object RoutingProgrammaticallyExample extends App {
val system = ActorSystem("RPE")
//#programmaticRoutingNrOfInstances
val router1 = system.actorOf(Props[ExampleActor1].withRouter(
RoundRobinRouter(nrOfInstances = 5)))
//#programmaticRoutingNrOfInstances
1 to 6 foreach { i ⇒ router1 ! Message1(i) }
//#programmaticRoutingRoutees
val actor1 = system.actorOf(Props[ExampleActor1])
val actor2 = system.actorOf(Props[ExampleActor1])
val actor3 = system.actorOf(Props[ExampleActor1])
val routees = Vector[ActorRef](actor1, actor2, actor3)
val router2 = system.actorOf(Props[ExampleActor1].withRouter(
RoundRobinRouter(targets = routees)))
//#programmaticRoutingRoutees
1 to 6 foreach { i ⇒ router2 ! Message1(i) }
}

403
akka-docs/scala/routing.rst
View file

@ -8,257 +8,284 @@ Routing (Scala)
Akka-core includes some building blocks to build more complex message flow handlers, they are listed and explained below:
Router
----------
------
A Router is an actor that routes incoming messages to outbound actors.
The router routes the messages sent to it to its underlying actors called 'routees'.
To use it you can either create a Router through the ``routerActor()`` factory method
Akka comes with four defined routers out of the box, but as you will see in this chapter it
is really easy to create your own. The four routers shipped with Akka are:
.. code-block:: scala
* `RoundRobinRouter <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Routing.scala#L173>`_
* `RandomRouter <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Routing.scala#L226>`_
* `BroadcastRouter <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Routing.scala#L284>`_
* `ScatterGatherFirstCompletedRouter <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Routing.scala#L330>`_
import akka.actor.Actor._
import akka.actor.Actor
import akka.routing.Routing._
To illustrate how to use the routers we will create a couple of simple actors and then use them in the
different router types.
//Our message types
case object Ping
case object Pong
Router usage
^^^^^^^^^^^^
//Two actors, one named Pinger and one named Ponger
//The actor(pf) method creates an anonymous actor and starts it
val pinger = actorOf(Props(new Actor { def receive = { case x => println("Pinger: " + x) } }))
val ponger = actorOf(Props(new Actor { def receive = { case x => println("Ponger: " + x) } }))
In this section we will describe how to use the different router types.
First we need to create some actors that will be used in the examples:
//A router that dispatches Ping messages to the pinger
//and Pong messages to the ponger
val d = routerActor {
case Ping => pinger
case Pong => ponger
}
.. includecode:: code/akka/docs/routing/RouterTypeExample.scala#printlnActor
d ! Ping //Prints "Pinger: Ping"
d ! Pong //Prints "Ponger: Pong"
and
Or by mixing in akka.routing.Router:
.. includecode:: code/akka/docs/routing/RouterTypeExample.scala#fibonacciActor
.. code-block:: scala
Here is the configuration file to instruct the routers how many instances of routees to create::
import akka.actor.Actor
import akka.actor.Actor._
import akka.routing.Router
//Our message types
case object Ping
case object Pong
class MyRouter extends Actor with Router {
//Our pinger and ponger actors
val pinger = actorOf(Props(new Actor { def receive = { case x => println("Pinger: " + x) } }))
val ponger = actorOf(Props(new Actor { def receive = { case x => println("Ponger: " + x) } }))
//When we get a ping, we dispatch to the pinger
//When we get a pong, we dispatch to the ponger
def routes = {
case Ping => pinger
case Pong => ponger
akka.actor.deployment {
/user/router {
nr-of-instances = 5
}
}
//Create an instance of our router, and start it
val d = actorOf(Props[MyRouter])
RoundRobinRouter
****************
Routes in a `round-robin <http://en.wikipedia.org/wiki/Round-robin>`_ fashion to its routees.
Code example:
d ! Ping //Prints "Pinger: Ping"
d ! Pong //Prints "Ponger: Pong"
.. includecode:: code/akka/docs/routing/RouterTypeExample.scala#roundRobinRouter
LoadBalancer
------------
A LoadBalancer is an actor that forwards messages it receives to a boundless sequence of destination actors.
Example using the ``loadBalancerActor()`` factory method:
When run you should see a similar output to this:
.. code-block:: scala
import akka.actor.Actor._
import akka.actor.Actor
import akka.routing.Routing._
import akka.routing.CyclicIterator
Received message '1' in actor $b
Received message '2' in actor $c
Received message '3' in actor $d
Received message '6' in actor $b
Received message '4' in actor $e
Received message '8' in actor $d
Received message '5' in actor $f
Received message '9' in actor $e
Received message '10' in actor $f
Received message '7' in actor $c
//Our message types
case object Ping
case object Pong
If you look closely to the output you can see that each of the routees received two messages which
is exactly what you would expect from a round-robin router to happen.
//Two actors, one named Pinger and one named Ponger
//The actor(pf) method creates an anonymous actor and starts it
RandomRouter
************
As the name implies this router type selects one of its routees randomly and forwards
the message it receives to this routee.
This procedure will happen each time it receives a message.
Code example:
val pinger = actorOf(Props(new Actor { def receive = { case x => println("Pinger: " + x) } }))
val ponger = actorOf(Props(new Actor { def receive = { case x => println("Ponger: " + x) } }))
.. includecode:: code/akka/docs/routing/RouterTypeExample.scala#randomRouter
//A load balancer that given a sequence of actors dispatches them accordingly
//a CyclicIterator works in a round-robin-fashion
val d = loadBalancerActor( new CyclicIterator( List(pinger,ponger) ) )
d ! Pong //Prints "Pinger: Pong"
d ! Pong //Prints "Ponger: Pong"
d ! Ping //Prints "Pinger: Ping"
d ! Ping //Prints "Ponger: Ping"
Or by mixing in akka.routing.LoadBalancer
When run you should see a similar output to this:
.. code-block:: scala
import akka.actor._
import akka.actor.Actor._
import akka.routing.{ LoadBalancer, CyclicIterator }
Received message '1' in actor $e
Received message '2' in actor $c
Received message '4' in actor $b
Received message '5' in actor $d
Received message '3' in actor $e
Received message '6' in actor $c
Received message '7' in actor $d
Received message '8' in actor $e
Received message '9' in actor $d
Received message '10' in actor $d
//Our message types
case object Ping
case object Pong
The result from running the random router should be different, or at least random, every time you run it.
Try to run it a couple of times to verify its behavior if you don't trust us.
//A load balancer that balances between a pinger and a ponger
class MyLoadBalancer extends Actor with LoadBalancer {
val pinger = actorOf(Props(new Actor { def receive = { case x => println("Pinger: " + x) } }))
val ponger = actorOf(Props(new Actor { def receive = { case x => println("Ponger: " + x) } }))
BroadcastRouter
***************
A broadcast router forwards the message it receives to *all* its routees.
Code example:
val seq = new CyclicIterator[ActorRef](List(pinger,ponger))
}
.. includecode:: code/akka/docs/routing/RouterTypeExample.scala#broadcastRouter
//Create an instance of our loadbalancer, and start it
val d = actorOf(Props[MyLoadBalancer])
d ! Pong //Prints "Pinger: Pong"
d ! Pong //Prints "Ponger: Pong"
d ! Ping //Prints "Pinger: Ping"
d ! Ping //Prints "Ponger: Ping"
Also, instead of using the CyclicIterator, you can create your own message distribution algorithms, there’s already `one <@http://github.com/jboner/akka/blob/master/akka-core/src/main/scala/routing/Iterators.scala#L31>`_ that dispatches depending on target mailbox size, effectively dispatching to the one that’s got fewest messages to process right now.
Example `<http://pastie.org/984889>`_
You can also send a 'Routing.Broadcast(msg)' message to the router to have it be broadcasted out to all the actors it represents.
When run you should see a similar output to this:
.. code-block:: scala
router ! Routing.Broadcast(PoisonPill)
Received message 'this is a broadcast message' in actor $f
Received message 'this is a broadcast message' in actor $d
Received message 'this is a broadcast message' in actor $e
Received message 'this is a broadcast message' in actor $c
Received message 'this is a broadcast message' in actor $b
As you can see here above each of the routees, five in total, received the broadcast message.
ScatterGatherFirstCompletedRouter
*********************************
The ScatterGatherFirstCompletedRouter will send the message on to all its routees as a future.
It then waits for first result it gets back. This result will be sent back to original sender.
Code example:
.. includecode:: code/akka/docs/routing/RouterTypeExample.scala#scatterGatherFirstCompletedRouter
When run you should see this:
.. code-block:: scala
The result of calculating Fibonacci for 10 is 55
From the output above you can't really see that all the routees performed the calculation, but they did!
The result you see is from the first routee that returned its calculation to the router.
Routers Explained
^^^^^^^^^^^^^^^^^
In the example usage above we showed you how to use routers configured with a configuration file but routers
can also be configured programatically.
This is an example of how to create a router and set the number of routees it should create:
.. includecode:: code/akka/docs/routing/RouterViaProgramExample.scala#programmaticRoutingNrOfInstances
You can also give the router already created routees as in:
.. includecode:: code/akka/docs/routing/RouterViaProgramExample.scala#programmaticRoutingRoutees
When you create a router programatically you define the number of routees *or* you pass already created routees to it.
If you send both parameters to the router *only* the latter will be used, i.e. ``nrOfInstances`` is disregarded.
*It is also worth pointing out that if you define the number of routees in the configuration file then this
value will be used instead of any programmatically sent parameters.*
Once you have the router actor it is just to send messages to it as you would to any actor:
.. code-block:: scala
router ! MyMsg
The router will apply its behavior to the message it receives and forward it to the routees.
Broadcast Messages
^^^^^^^^^^^^^^^^^^
There is a special type of message that will be sent to all routees regardless of the router.
This message is called 'Broadcast' and is used in the following manner:
.. code-block:: scala
router ! Broadcast("Watch out for Davy Jones' locker")
Only the actual message is forwarded to the routees, i.e. "Watch out for Davy Jones' locker" in the example above.
It is up to the routee implementation whether to handle the broadcast message or not.
Custom Router
^^^^^^^^^^^^^
You can also create your own router should you not find any of the ones provided by Akka sufficient for your needs.
In order to roll your own router you have to fulfill certain criteria which are explained in this section.
The router created in this example is a simple vote counter. It will route the votes to specific vote counter actors.
In this case we only have two parties the Republicans and the Democrats. We would like a router that forwards all
democrat related messages to the Democrat actor and all republican related messages to the Republican actor.
We begin with defining the class:
.. includecode:: ../../akka-actor-tests/src/test/scala/akka/routing/RoutingSpec.scala#crRouter
:exclude: crRoute
The next step is to implement the 'createRoute' method in the class just defined:
.. includecode:: ../../akka-actor-tests/src/test/scala/akka/routing/RoutingSpec.scala#crRoute
As you can see above we start off by creating the routees and put them in a collection.
Make sure that you don't miss to implement the line below as it is *really* important.
It registers the routees internally and failing to call this method will
cause a ``ActorInitializationException`` to be thrown when the router is used.
Therefore always make sure to do the following in your custom router:
.. includecode:: ../../akka-actor-tests/src/test/scala/akka/routing/RoutingSpec.scala#crRegisterRoutees
The routing logic is where your magic sauce is applied. In our example it inspects the message types
and forwards to the correct routee based on this:
.. includecode:: ../../akka-actor-tests/src/test/scala/akka/routing/RoutingSpec.scala#crRoutingLogic
As you can see above what's returned in the partial function is a ``List`` of ``Destination(sender, routee)``.
The sender is what "parent" the routee should see - changing this could be useful if you for example want
another actor than the original sender to intermediate the result of the routee (if there is a result).
For more information about how to alter the original sender we refer to the source code of
`ScatterGatherFirstCompletedRouter <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Routing.scala#L330>`_
All in all the custom router looks like this:
.. includecode:: ../../akka-actor-tests/src/test/scala/akka/routing/RoutingSpec.scala#CustomRouter
If you are interested in how to use the VoteCountRouter you can have a look at the test class
`RoutingSpec <https://github.com/jboner/akka/blob/master/akka-actor-tests/src/test/scala/akka/routing/RoutingSpec.scala>`_
Actor Pool
----------
An actor pool is similar to the load balancer is that it routes incoming messages to other actors. It has different semantics however when it comes to how those actors are managed and selected for dispatch. Therein lies the difference. The pool manages, from start to shutdown, the lifecycle of all delegated actors. The number of actors in a pool can be fixed or grow and shrink over time. Also, messages can be routed to more than one actor in the pool if so desired. This is a useful little feature for accounting for expected failure - especially with remoting - where you can invoke the same request of multiple actors and just take the first, best response.
An actor pool routes incoming messages to other actors. It has different semantics however when it comes to how those
actors are managed and selected for dispatch. Therein lies the difference. The pool manages, from start to shutdown,
the lifecycle of all delegated actors. The number of actors in a pool can be fixed or grow and shrink over time.
Also, messages can be routed to more than one actor in the pool if so desired. This is a useful little feature for
accounting for expected failure - especially with remoting - where you can invoke the same request of multiple
actors and just take the first, best response.
The actor pool is built around three concepts: capacity, filtering and selection.
Selection
^^^^^^^^^
All pools require a *Selector* to be mixed-in. This trait controls how and how many actors in the pool will receive the incoming message. Define *selectionCount* to some positive number greater than one to route to multiple actors. Currently two are provided:
All pools require a *Selector* to be mixed-in. This trait controls how and how many actors in the pool will
receive the incoming message. Define *selectionCount* to some positive number greater than one to route to
multiple actors. Currently two are provided:
* `SmallestMailboxSelector <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L133>`_ - Using the exact same logic as the iterator of the same name, the pooled actor with the fewest number of pending messages will be chosen.
* `RoundRobinSelector <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L158>`_ - Performs a very simple index-based selection, wrapping around the end of the list, very much like the CyclicIterator does.
* `SmallestMailboxSelector <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L148>`_ - Using the exact same logic as the iterator of the same name, the pooled actor with the fewest number of pending messages will be chosen.
* `RoundRobinSelector <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L184>`_ - Performs a very simple index-based selection, wrapping around the end of the list, very much like the CyclicIterator does.
Partial Fills
*************
When selecting more than one pooled actor, its possible that in order to fulfill the requested amount, the selection set must contain duplicates. By setting *partialFill* to **true**, you instruct the selector to return only unique actors from the pool.
When selecting more than one pooled actor, its possible that in order to fulfill the requested amount,
the selection set must contain duplicates. By setting *partialFill* to **true**, you instruct the selector to
return only unique actors from the pool.
Capacity
^^^^^^^^
As you'd expect, capacity traits determine how the pool is funded with actors. There are two types of strategies that can be employed:
* `FixedCapacityStrategy <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L268>`_ - When you mix this into your actor pool, you define a pool size and when the pool is started, it will have that number of actors within to which messages will be delegated.
* `BoundedCapacityStrategy <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L269>`_ - When you mix this into your actor pool, you define upper and lower bounds, and when the pool is started, it will have the minimum number of actors in place to handle messages. You must also mix-in a Capacitor and a Filter when using this strategy (see below).
* `FixedCapacityStrategy <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L346>`_ - When you mix this into your actor pool, you define a pool size and when the pool is started, it will have that number of actors within to which messages will be delegated.
* `BoundedCapacityStrategy <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L355>`_ - When you mix this into your actor pool, you define upper and lower bounds, and when the pool is started, it will have the minimum number of actors in place to handle messages. You must also mix-in a Capacitor and a Filter when using this strategy (see below).
The *BoundedCapacityStrategy* requires additional logic to function. Specifically it requires a *Capacitor* and a *Filter*. Capacitors are used to determine the pressure that the pool is under and provide a (usually) raw reading of this information. Currently we provide for the use of either mailbox backlog or active futures count as a means of evaluating pool pressure. Each expresses itself as a simple number - a reading of the number of actors either with mailbox sizes over a certain threshold or blocking a thread waiting on a future to complete or expire.
The *BoundedCapacityStrategy* requires additional logic to function. Specifically it requires a *Capacitor* and a *Filter*.
Capacitors are used to determine the pressure that the pool is under and provide a (usually) raw reading of this information.
Currently we provide for the use of either mailbox backlog or active futures count as a means of evaluating pool pressure.
Each expresses itself as a simple number - a reading of the number of actors either with mailbox sizes over a certain threshold
or blocking a thread waiting on a future to complete or expire.
Filtering
^^^^^^^^^
A *Filter* is a trait that modifies the raw pressure reading returned from a Capacitor such that it drives the adjustment of the pool capacity to a desired end. More simply, if we just used the pressure reading alone, we might only ever increase the size of the pool (to respond to overload) or we might only have a single mechanism for reducing the pool size when/if it became necessary. This behavior is fully under your control through the use of *Filters*. Let's take a look at some code to see how this works:
A *Filter* is a trait that modifies the raw pressure reading returned from a Capacitor such that it drives the
adjustment of the pool capacity to a desired end. More simply, if we just used the pressure reading alone,
we might only ever increase the size of the pool (to respond to overload) or we might only have a single
mechanism for reducing the pool size when/if it became necessary. This behavior is fully under your control
through the use of *Filters*. Let's take a look at some code to see how this works:
.. code-block:: scala
.. includecode:: code/akka/docs/routing/BoundedCapacitorExample.scala#boundedCapacitor
trait BoundedCapacitor
{
def lowerBound:Int
def upperBound:Int
.. includecode:: code/akka/docs/routing/CapacityStrategyExample.scala#capacityStrategy
def capacity(delegates:Seq[ActorRef]):Int =
{
val current = delegates length
var delta = _eval(delegates)
val proposed = current + delta
Here we see how the filter function will have the chance to modify the pressure reading to influence the capacity change.
You are free to implement filter() however you like. We provide a
`Filter <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L279>`_ trait that
evaluates both a rampup and a backoff subfilter to determine how to use the pressure reading to alter the pool capacity.
There are several sub filters available to use, though again you may create whatever makes the most sense for you pool:
if (proposed < lowerBound) delta += (lowerBound - proposed)
else if (proposed > upperBound) delta -= (proposed - upperBound)
* `BasicRampup <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L409>`_ - When pressure exceeds current capacity, increase the number of actors in the pool by some factor (*rampupRate*) of the current pool size.
* `BasicBackoff <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L426>`_ - When the pressure ratio falls under some predefined amount (*backoffThreshold*), decrease the number of actors in the pool by some factor of the current pool size.
* `RunningMeanBackoff <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L454>`_ - This filter tracks the average pressure-to-capacity over the lifetime of the pool (or since the last time the filter was reset) and will begin to reduce capacity once this mean falls below some predefined amount. The number of actors that will be stopped is determined by some factor of the difference between the current capacity and pressure. The idea behind this filter is to reduce the likelihood of "thrashing" (removing then immediately creating...) pool actors by delaying the backoff until some quiescent stage of the pool. Put another way, use this subfilter to allow quick rampup to handle load and more subtle backoff as that decreases over time.
delta
}
Example Usage
^^^^^^^^^^^^^
protected def _eval(delegates:Seq[ActorRef]):Int
}
trait CapacityStrategy
{
import ActorPool._
def pressure(delegates:Seq[ActorRef]):Int
def filter(pressure:Int, capacity:Int):Int
protected def _eval(delegates:Seq[ActorRef]):Int = filter(pressure(delegates), delegates.size)
}
Here we see how the filter function will have the chance to modify the pressure reading to influence the capacity change. You are free to implement filter() however you like. We provide a `Filter <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L279>`_ trait that evaluates both a rampup and a backoff subfilter to determine how to use the pressure reading to alter the pool capacity. There are several subfilters available to use, though again you may create whatever makes the most sense for you pool:
* `BasicRampup <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L308>`_ - When pressure exceeds current capacity, increase the number of actors in the pool by some factor (*rampupRate*) of the current pool size.
* `BasicBackoff <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L322>`_ - When the pressure ratio falls under some predefined amount (*backoffThreshold*), decrease the number of actors in the pool by some factor of the current pool size.
* `RunningMeanBackoff <https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/routing/Pool.scala#L341>`_ - This filter tracks the average pressure-to-capacity over the lifetime of the pool (or since the last time the filter was reset) and will begin to reduce capacity once this mean falls below some predefined amount. The number of actors that will be stopped is determined by some factor of the difference between the current capacity and pressure. The idea behind this filter is to reduce the likelihood of "thrashing" (removing then immediately creating...) pool actors by delaying the backoff until some quiescent stage of the pool. Put another way, use this subfilter to allow quick rampup to handle load and more subtle backoff as that decreases over time.
Examples
^^^^^^^^
.. code-block:: scala
class TestPool extends Actor with DefaultActorPool
with BoundedCapacityStrategy
with ActiveFuturesPressureCapacitor
with SmallestMailboxSelector
with BasicNoBackoffFilter
{
def receive = _route
def lowerBound = 2
def upperBound = 4
def rampupRate = 0.1
def partialFill = true
def selectionCount = 1
def instance(defaults: Props) = actorOf(defaults.withCreator(new Actor {def receive = {case n:Int =>
Thread.sleep(n)
counter.incrementAndGet
latch.countDown()}}))
}
.. code-block:: scala
class TestPool extends Actor with DefaultActorPool
with BoundedCapacityStrategy
with MailboxPressureCapacitor
with SmallestMailboxSelector
with Filter
with RunningMeanBackoff
with BasicRampup
{
def receive = _route
def lowerBound = 1
def upperBound = 5
def pressureThreshold = 1
def partialFill = true
def selectionCount = 1
def rampupRate = 0.1
def backoffRate = 0.50
def backoffThreshold = 0.50
def instance(defaults: Props) = actorOf(defaults.withCreator(new Actor {def receive = {case n:Int =>
Thread.sleep(n)
latch.countDown()}}))
}
Taken from the unit test `spec <https://github.com/jboner/akka/blob/master/akka-actor/src/test/scala/akka/routing/RoutingSpec.scala>`_.
.. includecode:: code/akka/docs/routing/ActorPoolExample.scala#testPool