无法编译带有未来列表的代码

Question

关于StackOverflow有一些关于如何处理期货列表的建议，但我想尝试我自己的方法。但是我无法编译以下代码

我有一份未来的清单。我想数一数他们中有多少人及格或不及格。我应该得到（2,1），我将其存储在一个元组中。我想采取的方法是遍历列表中的每个元素。列表的元素是Future[Int]。对于每个元素，我调用flatMap，它调用下一个递归循环（我假设如果调用flatMap，那么特定的未来将是成功的，因此我增加了传递计数）。类似地，我想在recover和increment fail count中调用下一个递归周期，但我在其中得到了编译错误。

import scala.concurrent._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.{Failure, Success, Try}
import scala.concurrent.duration._
import scala.language.postfixOps

object ConcurrencyExample extends App {

  type pass = Int
  type fail = Int

  val time = System.currentTimeMillis()

//use recursion to process each Future in the list 
  def segregate(l:List[Future[Int]]):Future[Tuple2[pass,fail]] = {
    def go(l:List[Future[Int]],t:Tuple2[pass,fail]):Future[Tuple2[pass,fail]] = {
        l match {
          case Nil => Future{t}
            //l is List of Future[Int]. flatMap each successful Future 
            //recover each failed Future
          case l::ls => {
            l flatMap (x => go(ls, (t._1 + 1, t._2)))
              **l.recover({ case e => go(ls, (t._1 + 1, t._2))})**//I get error here
          }
        }
    }
    go(l,(0,0))
  }

//hardcoded future
  val futures2: List[Future[Int]] = List(Future {
    1
  }, Future {
    2
  }, Future {
    throw new Exception("error")
  })


  val result = segregate(futures2)
  result onComplete {
    case Success(v) => println("pp:" + v)
    case Failure(v) => println("fp:" + v)
  }

  Await.result(result,1000 millis)
}

Answer 1

如果你看一下： T]（pf： PartialFunction[Throwable， U]）（implcit在执行器：scala.concurrent.ExecutionContext）：scala.concurrent.Future[U]"rel="no在执行器">文档，恢复的签名是：

def recover[U >: T](pf: PartialFunction[Throwable, U])(implicit executor: ExecutionContext): Future[U]

您在< code>l上调用的是< code>Future[Int]的< code>recover，因此< code>recover需要一个< code>U

但是，您正在再次调用go，其返回类型为Future[（通过，失败）]而不是

Answer 2

@evan058关于恢复的签名是正确的。但是您可以通过将恢复更改为恢复来修复您的程序。

recoverWith要像flatMap要映射一样进行恢复。

以下是完整的解决方案（略有样式改进）：

import scala.concurrent._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.{Failure, Success, Try}
import scala.concurrent.duration._
import scala.language.postfixOps

object ConcurrencyExample extends App {

  type pass = Int
  type fail = Int

  val time = System.currentTimeMillis()

  //use recursion to process each Future in the list
  def segregate[T](fs:List[Future[T]]):Future[(pass,fail)] = {
    def go(fs:List[Future[T]],r:Future[(pass,fail)]):Future[(pass,fail)] = fs match {
      case Nil => r
      case l::ls =>
        val fx = l.transform({_ => (1, 0)}, identity).recoverWith[(pass,fail)]({case _: Exception => Future(0, 1) })
        for (x <- fx; t <- r; g <- go(ls, Future(t._1+x._1,t._2+x._2))) yield g
    }
    go(fs,Future((0,0)))
  }

  //hardcoded future
  val futures2 = List(Future(1), Future(2), Future(throw new Exception("error")))    

  val result = segregate(futures2)
  result onComplete {
    case Success(v) => println(s"successes: ${v._1}, failures: ${v._2}")
    case Failure(v) => v.printStackTrace()
  }

  Await.result(result,1000 millis)
}