如何对含大量ID的Seq按批次并行处理并合并结果？

核心思路 #

将包含大量ID的序列按指定大小x拆分为多个批次，并行调用阻塞的getEntityFromID函数处理每个批次，最终将所有批次的结果合并为单个Seq[Entity]。

实现方式 #

方式一：基于Scala并行集合（简单快捷） #

利用Scala标准库的并行集合自动处理多线程并行，代码简洁易维护：

// 假设Entity类定义如下
case class Entity(id: Int, data: String)

// 模拟阻塞的实体加载函数
def getEntityFromID(ids: Seq[Int]): Iterable[Entity] = {
  // 模拟IO阻塞延迟
  Thread.sleep(100)
  ids.map(id => Entity(id, s"EntityData_$id"))
}

// 处理逻辑
val largeIdSeq: Seq[Int] = (1 to 50000).toSeq
val batchSize: Int = 100 // 自定义批次大小x

val finalEntities: Seq[Entity] = largeIdSeq
  .grouped(batchSize) // 拆分为大小为batchSize的批次
  .toList
  .par // 转换为并行集合，自动分配线程处理各批次
  .flatMap(getEntityFromID) // 调用阻塞函数并扁平化结果
  .seq // 转回普通Seq（根据需求可选）

注意点：

• 并行集合默认使用与CPU核心数匹配的线程池，适合CPU密集型任务；若为IO密集型场景，可能因线程不足导致阻塞等待，此时建议使用方式二。

方式二：基于Future（灵活控制并发） #

通过Future自定义线程池，更适配IO密集型场景，可灵活调整并发数：

import scala.concurrent.{Await, Future}
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext
import java.util.concurrent.Executors

// 自定义IO密集型线程池（比如设置10个线程）
implicit val ioExecutionContext: ExecutionContext = 
  ExecutionContext.fromExecutor(Executors.newFixedThreadPool(10))

case class Entity(id: Int, data: String)

def getEntityFromID(ids: Seq[Int]): Iterable[Entity] = {
  Thread.sleep(100)
  ids.map(id => Entity(id, s"EntityData_$id"))
}

// 处理逻辑
val largeIdSeq: Seq[Int] = (1 to 50000).toSeq
val batchSize: Int = 100

// 为每个批次创建Future
val batchFutures: Seq[Future[Iterable[Entity]]] = largeIdSeq
  .grouped(batchSize)
  .map(batch => Future(getEntityFromID(batch)))
  .toSeq

// 等待所有批次完成并合并结果
val finalEntities: Seq[Entity] = Await.result(
  Future.sequence(batchFutures).map(_.flatten),
  5.minutes // 根据实际业务设置超时时间
)

// 关闭自定义线程池（避免资源泄漏）
ioExecutionContext.shutdown()

优势：

• 可根据IO密集程度调整线程池大小，避免线程资源耗尽；
• 便于添加异常处理逻辑，比如通过recover处理单个批次加载失败的情况：

  val batchFutures: Seq[Future[Iterable[Entity]]] = largeIdSeq
    .grouped(batchSize)
    .map(batch => Future(getEntityFromID(batch)).recover {
      case e: Exception => 
        println(s"加载批次失败: $e")
        Iterable.empty[Entity]
    })
    .toSeq
  

批次大小x的选择建议 #

• 若目标服务有并发请求限制，需确保并行批次数量不超过限制；
• 过大的批次可能导致单个请求超时或服务压力过载；
• 过小的批次会增加请求次数，带来额外的网络/连接开销，建议根据实际场景测试调整。

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