Сложить влево

Простой foldLeft можно использовать для управления количеством фьючерсов, которые выполняются одновременно.

Во-первых, давайте создадим класс case под названием LazyFuture

case class LazyFuture[+A](f: Unit => Future[A]) {
  def apply() = f()
}

object LazyFuture {
  def apply[A](f: => A)(implicit ec: ExecutionContext): LazyFuture[A] = LazyFuture(_ => Future(f))

  def apply[A](f: => Future[A])(implicit ec: ExecutionContext): LazyFuture[A] = LazyFuture(_ => f)
}

LazyFuture немедленно останавливает запуск будущего

val list: List[LazyFuture[A]] = ...


list.grouped(concurFactor).foldLeft(Future.successful(List.empty[A])){ (r, c) =>
  val batch = Future.sequence(c.map(_()))
  batch.flatMap(values => r.map(rs => rs ++ values))
}

Измените concurFactor соответственно, чтобы запускать несколько фьючерсов одновременно.

concurFactor of 1 запустит одно будущее сразу

concurFactor of 2 будет запускать сразу два фьючерса

и так далее ...

def executeBatch[A](list: List[LazyFuture[A]])(concurFactor: Int) =
   list.grouped(concurFactor).foldLeft(Future.successful(List.empty[A])){ (r, c) =>
      val batch = Future.sequence(c.map(_()))
      r.flatMap(rs => batch.map(values => rs ++ values))
    }

Полный код

  case class LazyFuture[+A](f: Unit => Future[A]) {
    def apply() = f()
  }

  object LazyFuture {
    def apply[A](f: => A)(implicit ec: ExecutionContext): LazyFuture[A] = LazyFuture(_ => Future(f))

    def apply[A](f: => Future[A])(implicit ec: ExecutionContext): LazyFuture[A] = LazyFuture(_ => f)
  }

  def executeBatch[A](list: List[LazyFuture[A]])(concurFactor: Int)(implicit ec: ExecutionContext): Future[List[A]] =
    list.grouped(concurFactor).foldLeft(Future.successful(List.empty[A])) { (r, c) =>
      val batch = Future.sequence(c.map(_ ()))
      r.flatMap(rs => batch.map(values => rs ++ values))
    }

Ограничение контекста выполнения

Вы также можете ограничить вычислительные ресурсы, ограничив количество потоков в пуле выполнения. Но это решение не такое уж гибкое. Лично мне это не нравится.

val context: ExecutionContext = 
  ExecutionContext.fromExecutor(Executors.newFixedThreadPool(8))

Вы должны не забыть передать правильный контекст выполнения, который является неявным значением. Иногда мы не знаем, какое неявное значение входит в область действия. Это глючит

Предупреждение

Когда будущее построено, как показано ниже

val foo = Future {
     1 + 2
} // future starts executing

LazyFuture(foo) // Not a right way

foo уже запущен, и его нельзя контролировать.

Правильный способ построения LazyFuture

val foo = LazyFuture {
  1 + 2
}

или же

val foo = LazyFuture {
  Future {
   1 + 2
  }
}

Рабочий пример

package main

import scala.concurrent.{Await, ExecutionContext, Future}
import scala.concurrent.duration.Duration

object Main {

  case class LazyFuture[A](f: Unit => Future[A]) {
    def apply(): Future[A] = f()
  }

  object LazyFuture {
    def apply[A](f: => A)(implicit ec: ExecutionContext): LazyFuture[A] = LazyFuture(_ => Future(f))
    def apply[A](f: => Future[A]): LazyFuture[A] = LazyFuture(_ => f)
  }

  def executeBatch[A](list: List[LazyFuture[A]])(concurFactor: Int)
    (implicit ec: ExecutionContext): Future[List[A]] =
    list.grouped(concurFactor).foldLeft(Future.successful(List.empty[A])) { (r, c) =>
      val batch = Future.sequence(c.map(_ ()))
      r.flatMap(rs => r.map(values=> rs ++ values))
    }

  def main(args: Array[String]): Unit = {
    import scala.concurrent.ExecutionContext.Implicits.global


    val futures: Seq[LazyFuture[Int]] = List(1, 2, 3, 4, 5).map { value =>
      LazyFuture {
        println(s"value: $value started")
        Thread.sleep(value * 200)
        println(s"value: $value stopped")
        value
      }
    }
    val f = executeBatch(futures.toList)(2)
    Await.result(f, Duration.Inf)
  }

}

Параллелизм в Scala Futures управляется ExecutionContext. Обратите внимание, что фьючерсы начинают выполняться в контексте сразу после создания, поэтому ExecutionContext из Future.sequence не имеет особого значения. Вы должны предоставить соответствующий контекст при создании исходных фьючерсов из последовательности.

Контекст по умолчанию ExecutionContext.global (обычно импортируемый через import scala.concurrent.ExecutionContext.Implicits.global) использует столько потоков, сколько имеется ядер процессора, но он также может создавать множество дополнительных потоков для блокирующих задач, которые заключены в scala.concurrent.blocking. Обычно это желаемое поведение, но оно не подходит для вашей проблемы.

К счастью, вы можете использовать метод ExecutionContext.fromExecutor для обертывания пула потоков Java. Например:

import java.util.concurrent.Executors
import scala.concurrent.ExecutionContext

val context = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(10))
val seqOfFutures = Seq.fill(700)(Future { callRestApi() }(context))
val sequenceFuture = Future.sequence(seqOfFutures)(ExecutionContext.global)

Конечно, контекст также может быть предоставлен неявно:

implicit val context: ExecutionContext = 
  ExecutionContext.fromExecutor(Executors.newFixedThreadPool(10))
val seqOfFutures = Seq.fill(700)(Future { callRestApi() })
// This `sequence` uses the same thread pool as the original futures
val sequenceFuture = Future.sequence(seqOfFutures)