期货列表 - 发生异常时杀死所有期货

Question

我正在使用 scala 期货异步提交 1000 个作业。我还实现了一个由并发阻塞队列支持的 ThrottledExecutionContext，因此它一次最多只能运行 100 个作业，并将其余作业放入队列中。这是一个阻塞操作，因为它涉及在自身内部调用 3rd 方服务。当其中一个引发异常时，我需要重试整个操作（1000 个作业）或跳过整个批次。当某些期货仍在运行时，我无法重试。我有办法知道在任何时候有多少作业在第三方系统中运行 (spark)。因此，一旦我发现异常，我想首先杀死所有其余的期货，清空队列，等待第三方完成该批次的任何待处理作业，然后重试。那么有没有办法在一个异常时杀死所有期货？

我根据以下讨论尝试了 failFast，但它没有达到我的预期。我对Promise 还没有更好的理解。但似乎我们可以用Promise 控制Future 的未来！

Scala Future/Promise fast-fail pipeline

  var atomicnt = new AtomicInteger() // to track how many jobs were finished when exception occured

  def failFast[T](futures: Seq[Future[T]]): Future[Seq[T]] = {
    val promise = Promise[Seq[T]]
    futures.foreach{f => f.onFailure{case ex => promise.failure(ex)}}
    val res = Future.sequence(futures)
    promise.completeWith(res).future
  }

  def normalTask() =  {
    println("Starting normaltask")
    Thread.sleep(2000 + Random.nextInt(5000))
    if(Random.nextDouble() > 0.5) {
      println("Throwing random exception..")
      throw new RuntimeException("Random exception from normalTask")
    }
    atomicnt.getAndIncrement
    Thread.sleep(2000 + Random.nextInt(5000))
    println("Finished normaltask")
  }

  def testException() = {
    val rg = (0 until 500)
    val futures = rg.map(i =>{
      Future(normalTask)
    })
    val res = failFast(futures)
    Await.result(res, Duration.Inf) //blocking here to wait for all 500 to finish    
  }

  def batchProcessing() {
    
    try {
      println("Starting batchProcessing")
      testException()
      println("Exiting batchProcessing")      
    } catch {
      case t: Throwable => {
        println("Error in main")
        Thread.sleep(10000) //Here while waiting other futures are still running
        t.printStackTrace()
        // retry logic goes here based on failure or entire batch will be skipped 
      }
    }
    
  }

但是，当我在 batchProcessing 中捕获异常时，其他期货仍在运行。

我尝试进行并行处理的其他选项是使用似乎可行的并行集合。 IE。如果任何任务失败，则整个并行操作都会失败。但是，问题在于吞吐量不受我可用的 CPU 数量的限制。由于所有任务都在长时间运行，并且阻塞并行收集在那里感觉不正确。

Answer 1

内置的 scala Future 一旦启动就不能被中断。

似乎您需要像 monix Task 或 ZIO 这样可以很容易地中断和重试的东西。