In certain scenarios it is important to limit the number of concurrent requests to other services. For example, to avoid overwhelming the services and avoid performance degradation, or to maintain service level agreements. This is particularly important when streams are unbounded and the message rates are dynamic.

No matter the scenario, the Akka Streams API provides a seamless way to do this through back pressure applied upstream.

The following example shows how to batch elements, then asynchronously write the batched elements to a database, limiting the number of outstanding requests to only 10.

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

  def writeToDB(batch: Seq[Int])= Future {
    println(s"Writing ${batch.size} elements to the DB using thread '${Thread.currentThread().getName}'")
  }

  val rateLimitedGraph = Source(1 to 100000)
    .groupedWithin(100, 100.millis)
    .mapAsync(10)(writeToDB)
    .runWith(Sink.ignore)
    .onComplete(_ => system.terminate())

The above example preserves the order of the elements downstream, which can be important depending on the application. If downstream order of elements is not important the Akka Streams API provides mapAsyncUnordered.

For more on mapAsync see the Java or Scala documentation.
For more on mapAsyncUnordered see the Java or Scala documentation.