kenlz kenlz - 3 months ago 18
Scala Question

Difference between RoundRobinRouter and RoundRobinRoutinglogic

So I was reading tutorial about akka and came across this http://manuel.bernhardt.io/2014/04/23/a-handful-akka-techniques/ and I think he explained it pretty well, I just picked up scala recently and having difficulties with the tutorial above,

I wonder what is the difference between RoundRobinRouter and the current RoundRobinRouterLogic? Obviously the implementation is quite different.

Previously the implementation of RoundRobinRouter is

val workers = context.actorOf(Props[ItemProcessingWorker].withRouter(RoundRobinRouter(100)))


with processBatch

def processBatch(batch: List[BatchItem]) = {

if (batch.isEmpty) {
log.info(s"Done migrating all items for data set $dataSetId. $totalItems processed items, we had ${allProcessingErrors.size} errors in total")
} else {
// reset processing state for the current batch
currentBatchSize = batch.size
allProcessedItemsCount = currentProcessedItemsCount + allProcessedItemsCount
currentProcessedItemsCount = 0
allProcessingErrors = currentProcessingErrors ::: allProcessingErrors
currentProcessingErrors = List.empty

// distribute the work
batch foreach { item =>
workers ! item
}
}

}


Here's my implementation of RoundRobinRouterLogic

var mappings : Option[ActorRef] = None

var router = {
val routees = Vector.fill(100) {
mappings = Some(context.actorOf(Props[Application3]))
context watch mappings.get
ActorRefRoutee(mappings.get)
}
Router(RoundRobinRoutingLogic(), routees)
}


and treated the processBatch as such

def processBatch(batch: List[BatchItem]) = {

if (batch.isEmpty) {
println(s"Done migrating all items for data set $dataSetId. $totalItems processed items, we had ${allProcessingErrors.size} errors in total")
} else {
// reset processing state for the current batch
currentBatchSize = batch.size
allProcessedItemsCount = currentProcessedItemsCount + allProcessedItemsCount
currentProcessedItemsCount = 0
allProcessingErrors = currentProcessingErrors ::: allProcessingErrors
currentProcessingErrors = List.empty

// distribute the work
batch foreach { item =>
// println(item.id)
mappings.get ! item
}
}
}


I somehow cannot run this tutorial, and it's stuck at the point where it's iterating the batch list. I wonder what I did wrong.

Thanks

Answer

Actually I misunderstood the method, and found out the solution was to use RoundRobinPool as stated in http://doc.akka.io/docs/akka/2.3-M2/project/migration-guide-2.2.x-2.3.x.html

For example RoundRobinRouter has been renamed to RoundRobinPool or RoundRobinGroup depending on which type you are actually using.

from

val workers = context.actorOf(Props[ItemProcessingWorker].withRouter(RoundRobinRouter(100)))

to

val workers = context.actorOf(RoundRobinPool(100).props(Props[ItemProcessingWorker]), "router2")