CODEWITHSUNDEEP
scalaakka
blue-sky
blue-sky

Reputation: 53806

Limit number of messages sent within time interval

Using below code I'm attempting to limit the amount of messages send to an actor within a specified time frame. But the messages are not being throttled and are being sent as quickly as possible. The downstream actor just makes a http request to the Google home page.

The throttler code where I attempt to limit 3 messages to be sent within 3 seconds :

  val throttler: ActorRef =
    Source.actorRef(bufferSize = 1000, OverflowStrategy.dropNew)
      .throttle(3, 1.second)
      .to(Sink.actorRef(printer, NotUsed))
      .run()

How can I limit the number of messages sent within loop :

  for( a <- 1 to 10000){

    // Create the 'greeter' actors
    val howdyGreeter: ActorRef =
      system.actorOf(Greeter.props(String.valueOf(a), printer), String.valueOf(a))

    howdyGreeter ! RequestActor("RequestActor")
    howdyGreeter ! Greet
  }

to 3 per second ?

entire code : 

//https://developer.lightbend.com/guides/akka-quickstart-scala/full-example.html

import akka.NotUsed
import akka.stream.{OverflowStrategy, ThrottleMode}
import akka.stream.scaladsl.{Sink, Source}
import org.apache.http.client.methods.HttpGet
import org.apache.http.entity.StringEntity
import org.apache.http.impl.client.DefaultHttpClient
import net.liftweb.json._
import net.liftweb.json.Serialization.write
import org.apache.http.util.EntityUtils
//import akka.contrib.throttle.TimerBasedThrottler
import akka.actor.{Actor, ActorLogging, ActorRef, ActorSystem, Props}
import scala.concurrent.duration._
import akka.NotUsed
import akka.actor.ActorRef
import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.OverflowStrategy
import akka.stream.ThrottleMode
import akka.stream.scaladsl.Sink
import akka.stream.scaladsl.Source

object Greeter {
  def props(message: String, printerActor: ActorRef): Props = Props(new Greeter(message, printerActor))
  final case class RequestActor(who: String)
  case object Greet
}

class Greeter(message: String, printerActor: ActorRef) extends Actor {
  import Greeter._
  import Printer._

  var greeting = ""

  def receive = {
    case RequestActor(who) =>

      val get = new HttpGet("http://www.google.com")
      val response = (new DefaultHttpClient).execute(get)
//      val responseString = EntityUtils.toString(response.getEntity, "UTF-8")
//      System.out.println(responseString)

      greeting = String.valueOf(response.getStatusLine.getStatusCode)
      println("message is "+message)
//      greeting = message + ", " + who
    case Greet           =>
      printerActor ! Greeting(greeting)
  }
}

object Printer {
  def props: Props = Props[Printer]
  final case class Greeting(greeting: String)
}

class Printer extends Actor with ActorLogging {
  import Printer._

  def receive = {
    case Greeting(greeting) =>
      log.info("Greeting received (from " + sender() + "): " + greeting)
  }
}

object AkkaQuickstart extends App {
  import Greeter._
  // Create the 'helloAkka' actor system
  val system: ActorSystem = ActorSystem("helloAkka")

  // Create the printer actor,this is also the target actor
  val printer: ActorRef = system.actorOf(Printer.props, "printerActor")

  implicit val materializer = ActorMaterializer.create(system)

  val throttler: ActorRef =
    Source.actorRef(bufferSize = 1000, OverflowStrategy.dropNew)
      .throttle(3, 1.second)
      .to(Sink.actorRef(printer, NotUsed))
      .run()

  //Create a new actor for each request thread
  for( a <- 1 to 10000){

    // Create the 'greeter' actors
    val howdyGreeter: ActorRef =
      system.actorOf(Greeter.props(String.valueOf(a), printer), String.valueOf(a))

    howdyGreeter ! RequestActor("RequestActor")
    howdyGreeter ! Greet
  }

}

Upvotes: 0

Views: 155

Answers (2)

Roland Kuhn
Roland Kuhn

Reputation: 15472

An actor cannot influence what other actors do, in particular it has no control over who puts messages in its mailbox and when — this is how the actor model works. An actor only gets to decide what to do with the messages it finds in its mailbox, and over this it has full control. It can for example drop them, send back error replies, buffer them, etc.

If you want throttling and back-pressure, I recommend not using Actors at all for this part, but only use Akka Streams. The code that generates your request messages should be a Source, not a for-loop. Which source is most appropriate depends entirely on your real use-case, e.g. creating a stream from a strict collection with Source.from() or asynchronously pulling new elements out of a data structure with Source.unfoldAsync plus many more. Doing it this way ensures that the requests are only emitted when the time is right according to the downstream capacity or rate throttling.

Upvotes: 2

Levi Ramsey
Levi Ramsey

Reputation: 20551

It does not appear to me that you're actually using the throttler:

  val throttler: ActorRef =
    Source.actorRef(bufferSize = 1000, OverflowStrategy.dropNew)
      .throttle(3, 1.second)
      .to(Sink.actorRef(printer, NotUsed))
      .run()

But I don't see any messages being sent to throttler in your code: throttler will only throttle messages sent to throttler.

Upvotes: 1

Related Questions