scala - 如何限制Scala中未处理的 future 数量？

Question

如果有办法限制Scala中未处理的 future 数量，我将无法提供资金。
例如下面的代码:

import ExecutionContext.Implicits.global    
for (i <- 1 to N) {
  val f = Future {
    //Some Work with bunch of object creation
  }
}

如果N太大，最终将抛出OOM。
有没有一种方法可以通过类似队列的等待或异常来限制未处理的 future 以太币的数量？

Answer 1

因此，最简单的答案是，您可以创建一个ExecutionContext来阻止或限制超过一定限制的新任务的执行。参见this blog post。有关阻塞的Java ExecutorService的更加充实的示例，这里是an example。 [如果需要，您可以直接使用它，Maven Central上的库是here。]这包装了一些非阻塞的ExecutorService，您可以使用java.util.concurrent.Executors的工厂方法来创建它们。

将Java ExecutorService转换为Scala ExecutionContext只是ExecutionContext.fromExecutorService( executorService )。因此，使用上面链接的库，您可能会有类似以下的代码:

import java.util.concurrent.{ExecutionContext,Executors}
import com.mchange.v3.concurrent.BoundedExecutorService

val executorService = new BoundedExecutorService(
  Executors.newFixedThreadPool( 10 ), // a pool of ten Threads
  100,                                // block new tasks when 100 are in process
  50                                  // restart accepting tasks when the number of in-process tasks falls below 50
 )

implicit val executionContext = ExecutionContext.fromExecutorService( executorService )

// do stuff that creates lots of futures here...

如果您想要一个有限的 ExecutorService，它可以和整个应用程序一样长的时间，那就很好了。但是，如果您要在代码的本地化点中创建大量 future ，那么您将需要在完成处理后关闭 ExecutorService。我 define loan-pattern methods in Scala [ maven central]都创建上下文并在完成后将其关闭。该代码最终看起来像...

import com.mchange.sc.v2.concurrent.ExecutionContexts

ExecutionContexts.withBoundedFixedThreadPool( size = 10, blockBound = 100, restartBeneath = 50 ) { implicit executionContext =>
    // do stuff that creates lots of futures here...

    // make sure the Futures have completed before the scope ends!
    // that's important! otherwise, some Futures will never get to run
}

您可以使用实例来强制执行任务调度( ExecutorService -creating) Future来执行任务，而不是异步运行它，而不是使用彻底阻止 Thread的实例。您将使用 java.util.concurrent.ThreadPoolExecutor制作一个 ThreadPoolExecutor.CallerRunsPolicy。但是 ThreadPoolExecutor直接构建非常复杂。

所有这些的更新，更性感，更以Scala为中心的替代方案是 checkout Akka Streams作为 Future的替代方案，并通过“反压”防止 OutOfMemoryErrors并发执行。