scala 为什么 Future.sequence 并行而不是串行执行我的期货？

Why Future.sequence executes my futures in parallel rather than in series?

提问by sscarduzio

The word "sequence" means a seriesof actions one after the other.

“顺序”这个词是指一系列一个接一个的动作。

object Test {

  def main(args: Array[String]) {

    def producer() = {
      val list = Seq(
          future { println("startFirst"); Thread.sleep(3000); println("stopFirst") }, 
          future { println("startSecond"); Thread.sleep(1000); println("stopSecond") }
      )
      Future.sequence(list)
    }

   Await.result(producer, Duration.Inf)
  }
}

Therefore I expect this program to print: startFirst stopFirst startSecond stopSecond

因此我希望这个程序打印： startFirst stopFirst startSecond stopSecond

or even: startSecond stopSecond startFirst stopFirst

甚至： startSecond stopSecond startFirst stopFirst

but not (as it happens): startFirst startSecond stopSecond stopFirst

但不是（因为它发生）： startFirst startSecond stopSecond stopFirst

Why this method is not called Future.parallel()? And what should I use to guarantee that all futures in a Seqof futures are triggered serially (as opposed to in parallel) ?

为什么不调用此方法Future.parallel()？我应该用什么来保证 a Seqof futures 中的所有 期货都是串行触发的（而不是并行触发）？

回答by grotrianster

The futures are running concurrently because they have been started concurrently :). To run them sequentially you need to use flatMap:

期货同时运行，因为它们已同时启动:)。要按顺序运行它们，您需要使用 flatMap：

Future { println("startFirst"); 
         Thread.sleep(3000); 
         println("stopFirst") 
        }.flatMap{
         _ =>  Future { 
                       println("startSecond"); 
                       Thread.sleep(1000); 
                       println("stopSecond") 
               }
        }

Future.sequence just turns Seq[Future[T]] => Future[Seq[T]]which means gather results of all already started futures and put it in future .

Future.sequence 只是轮流Seq[Future[T]] => Future[Seq[T]]，这意味着收集所有已经启动的期货的结果并将其放入未来。

回答by eagle yuan

a little change to the original Future.sequence will make the future execution serialized:

对原来的 Future.sequence 稍作改动，将使未来的执行序列化：

def seq[A, M[X] <: TraversableOnce[X]](in: M[() => Future[A]])(implicit cbf: CanBuildFrom[M[()=>Future[A]], A, M[A]], executor: ExecutionContext): Future[M[A]] = {
    in.foldLeft(Future.successful(cbf(in))) {
       (fr, ffa) => for (r <- fr; a <- ffa()) yield (r += a)
    } map (_.result())
}

and your code will look like this:

您的代码将如下所示：

object Test {
def main(args: Array[String]) {

    def producer() = {
      val list = Seq(
          {() => future { println("startFirst"); Thread.sleep(3000); println("stopFirst") }}, 
          {() => future { println("startSecond"); Thread.sleep(1000); println("stopSecond") }}
      )
      FutureExt.seq(list)
    }

    Await.result(producer, Duration.Inf)
    }
}

which is very like your original code and with same result collection with the original Future.sequence()

这与您的原始代码非常相似，并且与原始 Future.sequence() 具有相同的结果集合

回答by Viktor Klang

Linearize:

线性化：

import scala.concurrent._
import scala.collection.mutable.Builder
import scala.collection.generic.CanBuildFrom
import language.higherKinds

/**
 * Linearize asynchronously applies a given function in-order to a sequence of values, producing a Future with the result of the function applications.
 * Execution of subsequent entries will be aborted if an exception is thrown in the application of the function.
 */
def linearize[T, U, C[T] <: Traversable[T]](s: C[T])(f: T => U)(implicit cbf: CanBuildFrom[C[T], U, C[U]], e: ExecutionContext): Future[C[U]] = {
  def next(i: Iterator[T], b: Builder[U, C[U]]): Future[C[U]] = if(!i.hasNext) Future successful b.result else Future { b += f(i.next()) } flatMap { b => next(i, b) }
  next(s.toIterator, cbf(s))
}

scala> linearize(1 to 100)(_.toString) foreach println

scala> Vector(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99, 100)

From: https://gist.github.com/viktorklang/3347939

来自：https: //gist.github.com/viktorklang/3347939

回答by mixel

If you have a sequence of some values and you want to map them to Future's and execute them in series:

如果您有一些值的序列，并且想要将它们映射到Future's 并按顺序执行它们：

import scala.concurrent.ExecutionContext.Implicits.global

implicit class SeqExtension[A](s: Seq[A]) {
  def foldLeftToFuture[B](initial: B)(f: (B, A) => Future[B])(implicit ec: ExecutionContext): Future[B] =
    s.foldLeft(Future(initial))((future, item) => future.flatMap(f(_, item)))

  def mapInSeries[B](f: A => Future[B])(implicit ec: ExecutionContext): Future[Seq[B]] =
    s.foldLeftToFuture(Seq[B]())((seq, item) => f(item).map(seq :+ _))
}

val stringsFuture: Future[Seq[String]] = Seq(1, 2, 3).mapInSeries[String](i => Future(i.toString))
val strings = Await.result(stringsFuture, Duration.Inf) // List("1", "2", "3")

回答by Gomes

or you can use for {..} yield block to get sequence of Future. Future.sequence convert Seq[Future] to Future[Seq]

或者您可以使用 for {..} yield 块来获取 Future 的序列。Future.sequence 将 Seq[Future] 转换为 Future[Seq]