使用带有 onComplete 的 Scala 期货列表进行异步处理以进行异常处理

Question

我正在尝试进行大量外部服务调用，每个调用都进行异常处理和有条件的进一步处理。我认为使用内部的 .onComplete 扩展这个很好的 (Asynchronous IO in Scala with futures) 示例会很容易，但似乎我不了解范围和/或期货。谁能指出我正确的方向吗？

#!/bin/bash
scala -feature $0 $@
exit
!#

import scala.concurrent.{future, blocking, Future, Await}
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
import scala.util.{Success, Failure}
import scala.language.postfixOps

val keylist = List("key1", "key2")

val myFuts: List[Future[String]] = keylist.map {
  myid => future {
    // this line simulates an external call which returns a future (retrieval from S3)
    val myfut = future { Thread.sleep(1); "START " + myid}

    var mystr = "This should have been overwritten"
    myfut.onComplete {
      case Failure(ex) => {
        println (s"failed with error: $ex")
        mystr = "FAILED"
      }
      case Success(myval) => {
        mystr = s"SUCCESS $myid: $myval"
        println (mystr)
      }
    }
    mystr
  }
}

val futset: Future[List[String]] = Future.sequence(myFuts)
println (Await.result(futset, 10 seconds))

在我的电脑 (Scala 2.10.4) 上打印：

SUCCESS key2: START key2
SUCCESS key1: START key1
List(This should have been overwritten, This should have been overwritten)

我想要（订单不重要）：

SUCCESS key2: START key2
SUCCESS key1: START key1
List(SUCCESS key2: START key2, SUCCESS key1: START key1)

Answer 1

完成时不会返回新的未来，它只是允许您在未来完成时做某事。因此，您的第一个 Future 块在执行 onComplete 之前返回，因此您将取回字符串的原始值。

我们可以做的是使用一个promise，返回另一个future，这个future由第一个future的结果完成。

  val keylist = List("key1", "key2")

  val myFuts: List[Future[String]] = keylist.map {
    myid => {
      // this line simulates an external call which returns a future (retrieval from S3)
      val myfut = Future {
        Thread.sleep(1); "START " + myid
      }
      var mystr = "This should have been overwritten"
      val p = Promise[String]()
      myfut.onComplete {
        case Failure(ex) =>
          println(s"failed with error: $ex")
          mystr = "FAILED"
          p failure ex
        case Success(myval) =>
          mystr = s"SUCCESS $myid: $myval"
          println(mystr)
          p success myval
      }
      p.future
    }
  }

  val futset: Future[List[String]] = Future.sequence(myFuts)
  println(Await.result(futset, 10 seconds))

我在这里询问的 mapAll 方法会非常方便： Map a Future for both Success and Failure

Answer 2

我会避免使用onComplete 并尝试使用它对可变变量执行副作用逻辑。相反，我会映射未来并将失败情况处理为返回不同的值。这是您的代码的略微修改版本，在Future 上使用map（通过理解），然后使用recover 处理失败情况。希望这就是您想要的：

val keylist = List("key1", "key2")

val myFuts: List[Future[String]] = keylist.map {myid => 

  // this line simulates an external call which returns a future (retrieval from S3)
  val myfut = future { Thread.sleep(1); "START " + myid}
  val result = for (myval <- myfut) yield {
    val res = s"SUCCESS $myid: $myval"
    println(res)
    res
  }
  result.recover{
    case ex => 
      println (s"failed with error: $ex")
      "FAILED"          
  }

}

val futset: Future[List[String]] = Future.sequence(myFuts)
println (Await.result(futset, 10 seconds))