Scala中Future数量超CPU核心数时无法异步执行的问题排查与解决

嘿，我来帮你拆解这个问题！你用Scala Future实现多线程时，发现任务超过CPU核心数就会分组执行，一组跑完另一组才启动，这其实是默认线程池的限制导致的，咱们一步步来解决：

你的代码与现象分析 #

先贴出你提供的代码（修正了转义符号）：

import scala.collection.mutable._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent._
import scala.concurrent.duration._
import scala.language.postfixOps
object FutureTest {
  def main(args: Array[String]) {
    val threads=10
    def ft(): Future[String] = Future {
      for (i <- 1 until 3) {
        Thread.sleep(1000)
        println(Thread.currentThread().getName + "\t" + i)
      }
      Thread.currentThread().getName + " end..."
    }
    var fs = Set[Future[String]]()
    for (j <- 1 until threads) {
      val f = ft
      f.onComplete { case _ => "Thread :" + j + " complete" }
      fs += f
    }
    fs.foreach(f => { Await.ready(f, Duration.Inf) })
  }
}

从你的终端输出能看到，前8个任务（对应你的CPU核心数，比如8核）先完成了两轮打印，第9个任务才开始执行——这明显是任务被排队调度了。

问题根因 #

你用的ExecutionContext.Implicits.global是Scala默认提供的线程池，它底层是ForkJoinPool，默认并行度等于你的CPU核心数（比如8核就同时跑8个任务）。当你创建的Future数量超过这个并行度时，多余的任务会被放到队列里，只有当正在运行的任务完成、线程空闲下来时，队列里的任务才会被调度执行，所以看起来就像“分组执行”。

另外还有个小细节：你在onComplete里只是返回了一个字符串，但没有打印，所以你看不到任务完成的提示，这个可以顺便修正。

修复方案 #

解决办法很简单：自定义一个并行度足够的ExecutionContext，让所有任务能同时启动。比如用固定线程池来实现：

import scala.collection.mutable._
import scala.concurrent._
import scala.concurrent.duration._
import scala.language.postfixOps
import java.util.concurrent.Executors

object FutureTest {
  // 自定义ExecutionContext，设置固定线程数为10，足够容纳你的9个任务
  implicit val customEc: ExecutionContextExecutor = 
    ExecutionContext.fromExecutor(Executors.newFixedThreadPool(10))

  def main(args: Array[String]) {
    val threads = 10
    def ft(): Future[String] = Future {
      for (i <- 1 until 3) {
        Thread.sleep(1000)
        println(Thread.currentThread().getName + "\t" + i)
      }
      Thread.currentThread().getName + " end..."
    }
    
    val fs = (1 until threads).map { j =>
      val f = ft()
      // 修正onComplete，添加打印
      f.onComplete { _ => println(s"Thread :$j complete") }
      f
    }.toSet

    // 用Future.sequence等待所有任务完成，比逐个Await更高效
    Await.result(Future.sequence(fs), Duration.Inf)
    // 最后记得关闭自定义线程池，避免资源泄漏
    customEc.shutdown()
  }
}

关键修改点： #

1. 自定义线程池：用Executors.newFixedThreadPool(10)创建了一个有10个线程的池，足够同时运行你创建的9个Future。
2. 优化等待逻辑：用Future.sequence(fs)把所有Future合并成一个，一次等待所有任务完成，比逐个Await.ready更简洁高效。
3. 修正onComplete：添加println，让你能看到任务完成的提示。
4. 关闭线程池：自定义线程池不会自动关闭，任务完成后手动调用shutdown()避免资源浪费。

修复后的效果 #

修改后运行代码，你会看到9个任务几乎同时启动，终端会一次性输出所有线程的1，1秒后再一次性输出所有线程的2，不会再出现分组执行的情况。

内容的提问来源于stack exchange，提问作者firefly