I am using akka 2.3.5 (including remoting module) to write a simple
proof-of-concept that utilizes consistent hashing routing with remote
deployed routees. My expectation is that the logic that spawns the router
actor will take care of deploying and instantiating the routee actors. In
my canonical example, all routees are configured for localhost. Below is
the code I wrote for this toy example:
> import java.lang.Thread.UncaughtExceptionHandler
>
> import java.util.concurrent.Executors
>
>
>> import akka.actor._
>
> import akka.pattern.ask
>
> import akka.remote.routing.RemoteRouterConfig
>
> import akka.routing.ConsistentHashingPool
>
> import akka.routing.ConsistentHashingRouter.ConsistentHashMapping
>
> import akka.util.Timeout
>
> import com.typesafe.config.ConfigFactory
>
>
>> import scala.concurrent.ExecutionContext
>
> import scala.concurrent.duration._
>
> import scala.util.{Failure, Success}
>
>
>> object ExampleRunner extends App {
>
>
>> Thread.currentThread().setUncaughtExceptionHandler(
>
> new UncaughtExceptionHandler {
>
> override def uncaughtException(t: Thread, e: Throwable): Unit =
>
> System.exit(1)
>
> })
>
>
>> implicit val responseTimeout = Timeout(3 seconds)
>
> val executorService = Executors.newFixedThreadPool(5)
>
> implicit val ec = ExecutionContext.fromExecutor(executorService)
>
> val as = ActorSystem(name = "example", config = ConfigFactory.load())
>
>
>> val addresses =
>
> Seq(
>
> AddressFromURIString("akka.tcp://[email protected]:5001"),
>
> AddressFromURIString("akka.tcp://[email protected]:5002"),
>
> AddressFromURIString("akka.tcp://[email protected]:5003"),
>
> AddressFromURIString("akka.tcp://[email protected]:5004"),
>
> AddressFromURIString("akka.tcp://[email protected]:5005"))
>
>
>> def hashMapping: ConsistentHashMapping = {
>
> case s => s
>
> }
>
>
>> val router =
>
> as.actorOf(
>
> props =
>
> RemoteRouterConfig(
>
> local =
>
> ConsistentHashingPool(
>
> nrOfInstances = addresses.size,
>
> virtualNodesFactor = 30,
>
> hashMapping = hashMapping),
>
> nodes = addresses)
>
> .props(PseudoWorker.props),
>
> name = "router")
>
>
>> print(">")
>
> for (line <-
>> Stream.continually(Console.readLine()).takeWhile(_.nonEmpty)) {
>
> router.ask(line).mapTo[String].onComplete {
>
> case Success(response) =>
>
> println(s"response = $response")
>
> print(">")
>
> case Failure(e) =>
>
> println(s"Send failed due to ${e.getMessage}")
>
> println(">")
>
> }
>
> }
>
>
>> println("Terminating")
>
> as.shutdown()
>
> executorService.shutdownNow()
>
> }
>
>
>> class PseudoWorker extends Actor with ActorLogging {
>
>
>> def receive = {
>
> case message =>
>
> log.info(s"Received $message")
>
> sender ! message
>
> }
>
> }
>
>
>> object PseudoWorker {
>
>
>> def props: Props = Props(new PseudoWorker)
>
> }
>
>
For completeness, here is the application.conf:
> akka {
actor {
provider = "akka.remote.RemoteActorRefProvider"
}
remote {
enabled-transports = ["akka.remote.netty.tcp"]
netty.tcp {
hostname = "127.0.0.1"
port = 2552
}
}
}
Running ExampleRunner produces the following output for each worker routee
actor:
> akka.remote.InvalidAssociation: Invalid address:
> akka.tcp://[email protected]:5002
> Caused by: akka.remote.transport.Transport$InvalidAssociationException:
> Connection refused: /127.0.0.1:5002
I would appreciate feedback to help me understand where I've misinterpreted
the routing <http://doc.akka.io/docs/akka/2.3.5/scala/routing.html>
documentation in setting up remote deployed routees.
Thank you,
Michael
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
