Scala：文件数据的顺序处理答案

【问题标题】：Scala: Sequential processing of file dataScala：文件数据的顺序处理
【发布时间】：2017-05-12 09:35:05
【问题描述】：

我有一个 csv 文件，我从中读取数据并填充我的数据库。我正在使用 scala 来执行此操作。我不想以并行方式触发数据库插入，而是希望以顺序方式执行插入（即一个接一个）。我不愿意在 for 循环中使用 Await。除了使用 await 还有其他方法吗？

P.S：我已将 csv 中的 1000 个条目读取到一个列表中，并在列表上循环以创建数据库插入

【问题讨论】：

你有异步方法插入数据库吗？如果不是这样，为什么不在循环中一一执行插入语句？

标签： scala future sequential

【解决方案1】：

假设您的数据库有某种 save(entity: T): Future[_] 方法，您可以使用 flatMap 折叠您的期货（或用于理解）：

def saveAll(entities: List[T]): Future[Unit] 
  entities.foldLeft(Future.successful(())){
    case (f, entity) => for {
        _ <- f
        _ <- save(entity)
      } yield ()
    }
  }

【讨论】：

【解决方案2】：

另一个选项是递归函数。不如foldLeft 简洁，但对某些人来说更具可读性。只需另一种选择供您考虑（假设save(entity: T): Future[R]：

def saveAll(entities: List[T]): Future[List[R]] = {
  entities.headOption match {
    case Some(entity) => 
      for {
        head <- save(entity)
        tail <- saveAll(entities.tail)
      } yield {
        head :: tail
      }
    case None =>
      Future.successful(Nil)
  }
}

如果您的save 方法允许您提供自己的ExecutionContext 即save(entity: T)(implicit ec: ExecutionContext): Future[R]，则另一种选择是同时触发Futures，但使用单线程执行上下文：

def saveAll(entities: List[T]): Future[List[R]] = {
  implicit ec = ExecutionContext.fromExecutionService(java.util.concurrent.Executors.newSingleThreadExecutor)
  Future.sequence(entities.map(save))
}

【讨论】：

需要关闭单线程执行器。我不建议创建这样的线程，而是需要调用者的 EC。