How to do Throttling in Akka Streams

When building a streaming application you may find the need to throttle the upstream so as to avoid exceeding a specified rate. Akka Stream's provides the capability to either fail the stream or shape it by applying back pressure. This is simply done by adding a throttle element and specifying the number of elements per time unit.

implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()
implicit val ec = system.dispatcher

val throttleGraph = Source(1 to 1000000)
    .map(n => s"I am number $n")
    .throttle(elements = 1, per = 1 second, maximumBurst = 1, mode = ThrottleMode.shaping)
    .runWith(Sink.foreach(println))
    .onComplete(_ => system.terminate())

Once the upper bound has been reached the parameter maximumBurst can be used to allow the client to send a burst of messages while still respecting the throttle. This is further exhibited in the below example.

implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()
implicit val ec = system.dispatcher

def writeToDB(batch: Seq[Int]): Future[Unit] = Future {
    println(s"Writing ${batch.size} elements to the DB using thread '${Thread.currentThread().getName}'")
}
val throttlerGraph2 = Source(1 to 1000000)
    .grouped(10)
    .throttle(elements = 10, per = 1 second, maximumBurst = 10, mode = ThrottleMode.shaping)
    .mapAsync(10)(writeToDB)
    .runWith(Sink.ignore)
    .onComplete(_ => system.terminate())

See more in the Java or the Scala documentation.