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

Question

我使用scala futures异步提交了1000份工作。我还实现了一个由并发阻塞队列支持的ThrottledExecutionContext，这样它一次最多只能运行100个作业，并将其余的放入队列中。这是一个阻塞操作，因为它涉及调用第三方服务本身。当其中一个抛出异常时，我需要重试整个操作(1000个作业)或者跳过整个批处理。当某些期货仍在运行时，我不能重试。我有办法知道在任何给定的时间点有多少作业在第三方系统中运行(< code>spark)。因此，一旦我捕捉到一个异常，我想首先杀死所有剩余的期货，清空队列，等待第三方完成该批处理的任何挂起的作业，然后重试。那么，有没有一种方法可以在一个例外中扼杀所有的未来？

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

Scala未来/promise快速失败管道

  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 
      }
    }
    
  }

然而，当我在批处理中捕捉到异常时，其他期货仍在运行。

我尝试并行处理的另一种选择是使用并行集合，这似乎有效。即。如果任何任务失败，则整个并行操作将失败。但是，问题在于吞吐量，它受我可用的CPU数量的限制。由于所有任务都长时间运行并且阻止并行收集感觉不是正确的方法。

Answer 1

内置scala＜code＞Future＜/code＞一旦启动就不能中断。

似乎您需要一些类似于monix任务或ZIO的东西，可以很容易地中断和重试。