Improve the Throttle for Akka(1)Local Cache

Improve the Throttle for Akka(1)Local Cache

1 Basic Idea of Akka
Start the Akka System, send messages.
package com.sillycat.throttle.demo
import akka.actor.{Actor, ActorSystem, Props}
class HelloActor extends Actor{

  def receive = {
    case "hello" => {
      println("hello!")
    }
    case _ => {
      println("huh?")
    }
  }
}

object HelloApp extends App {
  val system =  ActorSystem("HelloSystem")
  system.actorOf(Props[HelloActor], name="test1")
  system.actorOf(Props[HelloActor], name="test2")
  system.actorOf(Props[HelloActor], name="test3")

//  implicit val resolveTimeout = Timeout(5 seconds)
//  system.actorSelection("/user/test*").resolveOne().map { helloActor=>
//    println(helloActor)
//    helloActor ! "hello"
//    helloActor ! "bye"
//  }

  val helloActor = system.actorSelection("/user/test1")
      helloActor ! "hello"
      helloActor ! "bye"
  system.awaitTermination()
}

Simple Example with Router
package com.sillycat.throttle.demo

import akka.actor._
import akka.routing.FromConfig
import com.typesafe.config.ConfigFactory

object RoutingApp extends App{

  // A simple actor that prints whatever it receives
  class Printer extends Actor {
    def receive = {
      case x => println(self.path + " saying " + x)
    }
  }

  class Shooter extends Actor {
    def receive = {
      case x => println(self.path + " shouting " + x)
    }
  }

  val system =  ActorSystem("RoutingSystem", ConfigFactory.load())
  val router1: ActorRef = system.actorOf(Props[Printer].withRouter(FromConfig()), name = "Router1")
  // These three messages will be sent to the printer immediately
  router1 ! "11"
  router1 ! "12"
  router1 ! "13"
  // These two will wait at least until 1 second has passed
  router1 ! "14"
  router1 ! "15"
  println(" Router 1 " + router1.path)

  val router2: ActorRef = system.actorOf(Props[Shooter].withRouter(FromConfig()), name = "Router2")
  router2 ! "21"
  router2 ! "22"

  val router3: ActorSelection = system.actorSelection("/user/Router2")
  router3 ! "23"

  println(" Router 2 " + router2.path)

  system.shutdown()

}

2 Throttler Based on Local Cache
Try to implement this solution.
  //FUNCTION LIMIT_API_CALL(ip)
  //  ts = CURRENT_UNIX_TIME()
  //  keyname = ip+":"+ts
  //  current = GET(keyname)
  //  IF current != NULL AND current > 10 THEN
  //    ERROR "too many requests per second"
  //  ELSE
  //    MULTI
  //      INCR(keyname,1)
  //      EXPIRE(keyname,10)
  //    EXEC
  //      PERFORM_API_CALL()
  //END

I first build one implementation on top of guava local cache.
package actors.throttle

import akka.actor.{Actor, ActorRef}
import com.sillycat.util.IncludeLogger
import services.LocalCache
import utils.IncludeDateTimeUtil

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._

class LocalCacheBasedThrottler(var rate: Rate,var target: ActorRef) extends Actor with IncludeLogger with IncludeDateTimeUtil{

  val throttleKey = "THROTTLE_"

  def receive = {
    case msg: MessageTick => {
      val msgKey = msg.key
      val realMsg = msg.msg
      val counter = msg.count

      val limitCalls = rate.numberOfCalls
      val timeWindows = rate.duration

      val timeKey = convertCurrentTime2Key(timeWindows)
      val key = throttleKey + timeKey + msgKey

      LocalCache.throttleBucket.getIfPresent(key) match {
        case count:java.lang.Integer if count >= limitCalls => {
          //delay random and tick self
          LocalCache.throttleBucket.put(key, count + counter)
          val delay = calculateDelay(count + counter, limitCalls, timeWindows)
          //tick to self within the delay
          context.system.scheduler.scheduleOnce(delay second, self, msg)
        }
        case count:java.lang.Integer => {
          //count + 1
          LocalCache.throttleBucket.put(key, count + counter)
          //pass the ticket
          target ! realMsg
        }
        case _ => {
          //init the count
          LocalCache.throttleBucket.put(key, new Integer(counter))
          //pass the ticket
          target ! realMsg
        }
      }

    }
    case _ => {
      logger.error("Received a message I don't understand.")
    }
  }

}

case class Rate(val numberOfCalls: Int, val duration: Int)

case class MessageTick( key:String,  msg:Any, count: Int = 1)

The Local Cache Classes
package services

import java.util.concurrent.TimeUnit

import com.google.common.cache.CacheBuilder
import models.CountStep

object LocalCache {

  val builderThrottle = CacheBuilder.newBuilder().expireAfterWrite(60, TimeUnit.SECONDS)
  val throttleBucket = builderThrottle.build[java.lang.String, java.lang.Integer]()

  val builderMsg = CacheBuilder.newBuilder().expireAfterWrite(30, TimeUnit.MINUTES)
  val msgBucket = builderMsg.build[java.lang.String, CountStep]()

}

Here is how to use it.
    Akka.system.actorOf(Props(classOf[LocalCacheBasedThrottler],
      Rate(4, 15),
      contextIOActor),
      name = "context-io-throttler")

References:
http://sillycat.iteye.com/blog/2258226
http://letitcrash.com/post/28901663062/throttling-messages-in-akka-2
https://github.com/hbf/akka-throttler
http://sillycat.iteye.com/blog/1553508
http://doc.akka.io/api/akka/2.3.4/index.html#akka.contrib.throttle.TimerBasedThrottler
http://redis.io/commands/incr

http://doc.akka.io/docs/akka/snapshot/scala/routing.html