Scala使用Future.sequence执行批量异步任务出现OOM异常如何优化？

你遇到的OOM问题主要来自两个不合理的设计：

• 全量结果缓存：Future.sequence会等待所有异步任务完成，把近2000个ResultEntry全部加载到内存后再分组求最小值，如果Data对象本身内存占用较高，很容易触发内存溢出。
• 无限制并发：直接给所有计算任务包Future + blocking，如果使用Scala默认的全局ExecutionContext，blocking会无限制创建新线程，过多的线程栈+线程本地存储也会占用大量内存。

核心逻辑优化：增量计算最小值 #

你最终只需要每个目的地的最低分结果，完全不需要存储所有计算结果，直接在每个任务完成后更新对应目的地的当前最小值即可，内存占用直接从O(总任务数)降到O(目的地数量)，这里只有4个目的地，内存开销可以忽略。

并发度控制：设置合理的并行上限 #

既要保证足够的并发吞吐，又要避免线程过多导致的内存溢出，你可以根据任务类型（CPU密集/IO密集）设置并行度，比如IO密集型任务设为CPU核心数*4，CPU密集型任务设为等于CPU核心数即可。

优化后代码实现 #

import java.time.LocalDate
import java.util.concurrent.Executors
import java.util.concurrent.atomic.AtomicReference
import scala.concurrent.{Await, ExecutionContext, Future}
import scala.concurrent.duration.Duration

// 你的原有定义
case class Data(score: Double)
case class ResultEntry(destination: String, from: LocalDate, to: LocalDate, data: Data)
object SnelDwight {
  def calculateVacationScore(dest: String, from: LocalDate, to: LocalDate): Data = ??? // 保留原有实现
}

object VacationScoreCalc {
  def main(args: Array[String]): Unit = {
    // 自定义固定大小线程池控制并发度
    val parallelism = Runtime.getRuntime.availableProcessors() * 4
    implicit val ec: ExecutionContext = ExecutionContext.fromExecutor(Executors.newFixedThreadPool(parallelism))

    val destinations = Seq("New York", "Scranton", "Stamford", "Jamaica")
    val dates = 0.to(30).map(v => LocalDate.of(2021, 1, 1).plusDays(v))

    // 初始化每个目的地的最小值存储，用原子引用保证线程安全
    val minResults = destinations.map(dest => dest -> new AtomicReference[ResultEntry]()).toMap

    // 生成所有计算任务
    val allTasks = for {
      destination <- destinations
      fromDate <- dates
      toDate <- dates if toDate.isAfter(fromDate)
    } yield Future {
      val score = SnelDwight.calculateVacationScore(destination, fromDate, toDate)
      val entry = ResultEntry(destination, fromDate, toDate, score)
      // 原子更新当前目的地的最小值
      val currentMinRef = minResults(destination)
      var updated = false
      while (!updated) {
        val existingMin = currentMinRef.get()
        if (existingMin == null || entry.data.score < existingMin.data.score) {
          updated = currentMinRef.compareAndSet(existingMin, entry)
        } else {
          updated = true
        }
      }
    }

    // 等待所有任务完成
    Await.result(Future.sequence(allTasks), Duration.Inf)

    // 输出结果
    minResults.foreach { case (dest, minRef) =>
      println(s"${dest} : ${minRef.get()}")
    }

    // 关闭线程池
    ec.asInstanceOf[{def executor: java.util.concurrent.ExecutorService}].executor.shutdown()
  }
}

可选优化点 #

• 如果calculateVacationScore是CPU密集型计算，把并行度调整为等于CPU核心数即可，过高的并发反而会因为线程上下文切换降低性能。
• 如果Data对象非常大，可以只存储最小score和对应的日期，不需要存储整个Data对象，进一步降低内存开销。

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