scala 未来处理深度优先而不是广度优先

Question

我有一个大的计算大致基于以下模式：

def f1(i:Int):Int = ???
def f2(i:Int):Int = ???

def processA(l: List[Int]) = 
  l.map(i => Future(f1(i)))

def processB(l: List[Int]) = {
  val p = processA(l)
  p.map(fut => fut.map(f2))
}

def main() = {
  val items = List( /* 1k to 10k items here */ )
  val results = processB(items)
  results.map(_.onComplete ( ... ))
}

我遇到的问题，如果我的理解是正确的，就是处理是广度优先的。 ProcessA 启动数千个 Future，然后 processB 将在 processA 完成后处理的数千个新 Future 入队。 onComplete 回调将很晚才开始触发...

我想把这个深度优先：processA 的几个 Futures 开始，然后 processB 从那里继续，而不是切换到队列中的其他东西。

可以在 vanilla scala 中完成吗？我应该转向一些可以替代 Futures() 和 ThreadPools 的库吗？

编辑：更详细一点。正如答案中所建议的那样，重写为f1 andThen f2，目前是不可行的。实际上，processA and B 正在做很多其他事情（包括副作用）。 processB 依赖于 ProcessA 的事实是私有的。如果暴露会破坏 SoC。

编辑 2：我想我会放宽一点“原版”约束。有人建议 Akka 流会有所帮助。我目前正在查看 scalaz.Task：任何人的意见？

Answer 1

我不是 100% 确定我理解了这个问题，因为 processB (f2) 在 processA (f1) 的结果之上运行，您不能在尚未由 f1 计算的值上调用 f2，所以我的回答是基于以下假设：

• 您想限制进行中的工作
• 您想在f1 之后立即执行f2

所以这里有一个解决方案：

import scala.concurrent._
def process(noAtATime: Int, l: List[Int])(transform: Int => Int)(implicit ec: ExecutionContext): Future[List[Int]] = {
  // define an inner async "loop" to process one chunk of numbers at a time
  def batched(i: Future[Iterator[List[Int]]], result: List[List[Int]]): Future[List[Int]] =
    i flatMap { it =>
      // if there are more chunks to process
      // we process all numbers in the chunk as parallel as possible,
      // then combine the results into a List again, then when all are done,
      // we recurse via flatMap+batched with the iterator
      // when we have no chunks left, then we un-chunk the results
      // reassemble it into the original order and return the result
      if(it.hasNext) Future.traverse(it.next)(n => Future(transform(n))).flatMap(re => batched(i, re :: result))
      else Future.successful(result.reverse.flatten) // Optimize this as needed
    }
  // Start the async "loop" over chunks of input and with an empty result
  batched(Future.successful(l.grouped(noAtATime)), List.empty)
}


scala> def f1(i: Int) = i * 2 // Dummy impl to prove it works
f1: (i: Int)Int

scala> def f2(i: Int) = i + 1 // Dummy impl to prove it works
f2: (i: Int)Int

scala> process(noAtATime = 100, (1 to 10000).toList)(n => f2(f1(n)))(ExecutionContext.global)
res0: scala.concurrent.Future[List[Int]] = Future(<not completed>)

scala> res0.foreach(println)(ExecutionContext.global)

scala> List(3, 5, 7, 9, 11, 13, 15, 17, 19, 21, 23, 25, 27, 29, 31, 33, 35, 37, 39, 41, 43, 45, 47, 49, 51, 53, 55, 57, 59, 61, 63, 65, 67, 69, 71, 73, 75, 77, 79, 81, 83, 85, 87, 89, 91, 93, 95, 97, 99, 101, 103, 105, 107, 109, 111, 113, 115, 117, 119 …

如果您愿意并且能够使用更适合当前问题的库，have a look at this reply

Answer 2

您的问题最好用流表示。作业进入流并被处理，背压用于确保一次只完成有限数量的工作。在 Akka 流中，它看起来像这样：

Source(items)
  .mapAsync(4)(f1)
  .mapAsync(4)(f2)
  .<whatever you want to do with the result>

需要仔细选择并行度以匹配线程池大小，但这将确保通过f1 的平均次数将等于通过f2 的平均次数。

Answer 3

这有什么问题？

listOfInts.par.map(f1 andThen f2)

Answer 4

从异步计算的角度来看，从问题中不清楚是否需要将 f1 和 f2 分离为 processA 和 processB - f1 的结果始终且仅传递给 f2，可以这样做作为单个计算“f1 andThen f2”的一部分，并且在单个 Future 中更简单。

如果是这种情况，那么问题就简化为“如何在可能很大的输入上运行异步计算，从而限制正在运行的、衍生的 Futures”：

import scala.concurrent.Future
import java.util.concurrent.Semaphore
import scala.concurrent.ExecutionContext.Implicits.global

val f: (Int) => Int = i => f2(f1(i))

def process(concurrency: Int, input: List[Int], f: Int => Int): Future[List[Int]] = {
  val semaphore = new Semaphore(concurrency)
  Future.traverse(input) { i =>
    semaphore.acquire()
    Future(f(i)).andThen { case _ => semaphore.release() }
  }
}

Answer 5

可能是这样的：

items.foreach { it => processB(Seq(it)).onComplete(...) }

当然，如果您的f2 比f1 重得多，这也无济于事。在这种情况下，您将需要一些更明确的协调：

val batchSize = 10
val sem = new Semaphore(batchSize) 
items.foreach { it => 
   sem.acquire
   processB(Seq(It))
    .andThen { case _ => sem.release }
    .onComplete { ... }
}