Scala：結合所有成功和錯誤的未來串列

我有一個 Future 串列，每個 Future 都以值串列或失敗完成。我正在嘗試以捕獲所有可用值并記錄錯誤的方式組合這些期貨。這是我的嘗試：

val individualFutures:  List[Future[Seq[Element]]] = ....
val allElements: Future[List[Element]] = Future.foldLeft(individualFutures)(List[Element]())((acc, 
     elements) => acc    elements)
Try(Await.result(allElements, Duration.Inf)) match {
      case Success(elements) => ....
      case Failure(error)    => ....log
}

但我認為這不是實作我的目標的正確方法。我想一旦 future 完成并出現錯誤，它就會停在那里。

我怎樣才能達到同樣的效果？

uj5u.com熱心網友回復：

也許您可以通過回傳一個空串列并記錄失敗原因來遍歷串列并從 throwable 中恢復？最后你會展平整個序列。

  val allElements: Future[List[Element]] = Future.traverse(individualFutures) {
    _.recover {
      case throwable: Throwable =>
        println(throwable.getMessage)
        Nil
    }
  }.map(_.flatten)

uj5u.com熱心網友回復：

我稍微擴展了答案，但我認為您會發現它很有用。

您可以執行以下操作：

1. 創建一個輔助物件，用于聚合邏輯和記錄故障，您也可能希望獲得該組件的結果，該結果將告知是否發生故障。
2. 使用這個輔助物件來計算

幫手：

object FutureSupport extends StrictLogging {
  
  def aggregateSuccessSequence[In](futures: List[Future[In]], logOnFailure: Boolean = false)(implicit ec: ExecutionContext): Future[AggregateResult[In]] = {
    val futureTries = futures.map(_.map(Success(_)).recover {
      case NonFatal(ex) =>
        if (logOnFailure) {
          logger.warn(s"FutureSupport: Failure occurred on aggregateSuccessSequence", ex)
        }
        Failure(ex)
    })
    Future.sequence(futureTries).map {
      _.foldRight(AggregateResult.empty[In]) {
        case (curr, acc) =>
          curr match {
            case Success(res) => AggregateResult(acc.failureOccurred, res :: acc.outs)
            case Failure(_) => AggregateResult(failureOccurred = true, acc.outs)
          }
      }
    }
  }

  def aggregateSuccessTraversal[In, Out](source: List[In])(f: In => Future[Option[Out]])(logOnFailure: Boolean = false)(implicit ec: ExecutionContext): Future[AggregateResult[Out]] = {
    if (source.isEmpty) {
      Future.successful(AggregateResult.empty[In])
    }

    Future
      .traverse(source)(in => f(in).map(Success(_)).recover {
        case NonFatal(ex) =>
          if (logOnFailure) {
            logger.warn(s"FutureSupport: Failure occurred on aggregateSuccessTraversal", ex)
          }
          Failure(ex)
      })
      .map {
        _.foldRight(AggregateResult.empty[Out]) {
          case (curr, acc) =>
            curr match {
              case Failure(_) => acc.copy(failureOccurred = true)
              case Success(opt) => opt match {
                case Some(success) => acc.copy(outs = success :: acc.outs)
                case None => acc
              }
            }
        }
      }
  }

  case class AggregateResult[Out](failureOccurred: Boolean, outs: List[Out])
  object AggregateResult {
    def empty[Out]: AggregateResult[Out] = AggregateResult(failureOccurred = false, Nil)
  }
}

用法：

aggregateSuccessSequence(List(Future.successful(1), Future.failed(new Exception))).map {
      case AggregateResult(failureOccurred, outs) if  failureOccurred => ...
      case AggregateResult(failureOccurred, outs) => ... 
}

您可以使用相同的遍歷

標籤：斯卡拉未来

上一篇：如何根據bin范圍值將scala集合分類為子集

下一篇：SBT二進制不兼容問題scala-parser-combinators