Scala使用期货进行并行网络调用

问题描述 投票:0回答:1

我是Scala的新手,我有一个方法,它可以从给定的文件列表中读取数据,并使用数据,并将响应写入文件。

listOfFiles.map { file =>
  val bufferedSource = Source.fromFile(file)
  val data = bufferedSource.mkString
  bufferedSource.close()
  val response = doApiCall(data)  // time consuming task
  if (response.nonEmpty) writeFile(response, outputLocation)
}

上述方法,在网络通话期间花费了太多时间,因此尝试使用并行处理以减少时间。

所以我尝试包装代码块,这会花费更多时间,但是程序很快结束并不会产生任何输出,如上面的代码。

import scala.concurrent.ExecutionContext.Implicits.global

listOfFiles.map { file =>
  val bufferedSource = Source.fromFile(file)
  val data = bufferedSource.mkString
  bufferedSource.close()
  Future {
    val response = doApiCall(data) // time consuming task
    if (response.nonEmpty) writeFile(response, outputLocation)
  }
}

,如果您有任何建议,将会很有帮助。(我也尝试使用“ par”,效果很好,我正在探索“票面价格”以外的其他选项,并使用“ akka”,“ cats”等框架)

scala parallel-processing future
1个回答
0
投票

基于Jatin,而不是使用包含守护进程线程的默认执行上下文

import scala.concurrent.ExecutionContext.Implicits.global

使用非守护线程定义执行上下文

implicit val nonDeamonEc = ExecutionContext.fromExecutor(Executors.newCachedThreadPool)

也可以像这样使用Future.traverseAwait

val resultF = Future.traverse(listOfFiles) { file =>
  val bufferedSource = Source.fromFile(file)
  val data = bufferedSource.mkString
  bufferedSource.close()
  Future {
    val response = doApiCall(data) // time consuming task
    if (response.nonEmpty) writeFile(response, outputLocation)
  }
}

Await.result(resultF, Duration.Inf)

[traverseList[Future[A]]转换为Future[List[A]]

© www.soinside.com 2019 - 2024. All rights reserved.