如果花费的时间太长，则取消对并行集合的操作 - Scala

Question

是否可以取消对并行 Scala 集合的操作？例如，如果我这样做 myList.par.map(...)，是否可以稍后取消它，只获取已经处理过的元素？

Answer 1

可能无法取消对列表的操作，因为 Scala 库中没有 API 可以执行此操作。但是你有一些选择来完成这件事。例如，如果您的 List 是一个流，并且该流被包装在这样的 Future 中：

scala> import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.ExecutionContext.Implicits.global

scala> val stream = scala.concurrent.Future { 1 #:: 2 #:: 3 #:: Stream.empty }
stream: scala.concurrent.Future[scala.collection.immutable.Stream[Int]] = Success(Stream(1, ?))

您可以随时取消 Future - 基于一些超时！

另一种选择是使用 Observable 和 Observer（反应式），您可以将 List 建模为 Observable 并使用 Observer 订阅此 Observable。然后您可以随时取消订阅者！如果您对此感兴趣，请查看 Monix 库！

Answer 2

我没有找到内置答案 - 我决定只设置一个时间限制，如果时间超过阈值则停止迭代运行。

val MAX_TIME: Long = 120.minutes.toMillis
val start = System.currentTimeMillis()
val isTimeRemaining = () => (System.currentTimeMillis() - start) <= MAX_TIME

val processor:Seq[R] = (i:T) => {
  if (isTimeRemaining())
     //do some processing here
  else
    Nil
}

val results:Seq[T] = Random.shuffle(instances)
                      .par
                     .flatMap(processor)
                     .filterNot(_.isEmpty)
                     .toList

如果花费的时间太长，则取消对并行集合的操作 - Scala

Cancelling an operation on parallel collections if it takes too long - Scala

scala

concurrency

scala-collections