具有错误累积的异步处理的函数签名答案

【问题标题】：Function signature for async processing with errors accumulation具有错误累积的异步处理的函数签名
【发布时间】：2020-04-08 08:35:03
【问题描述】：

假设我有一个函数fab: A => Future[B]。现在我需要编写新函数foo 来处理Seq[A] 并累积所有错误。这就是我不能使用 Future.traverse 的原因，因为它“快速失败”并且不会累积错误。

foo 接收 Seq[A] 并应返回 Future。客户端应该获得B 或输入Seq[A] 的每个元素的异常。这个函数的签名是什么？

【问题讨论】：

标签： scala concurrency future

【解决方案1】：

要根据需要定义foo，请考虑在将fab 应用于输入列表的各个元素之后，在map/recover 之上使用Future.sequence，如下所示：

import scala.concurrent.{ Future, ExecutionContext }

def foo[A, B](ls: List[A])(fab: A => Future[B])(implicit ec: ExecutionContext):
    Future[List[Either[Throwable, B]]] =
  Future.sequence(ls.map(fab).map(_.map(Right(_)).recover{ case e => Left(e) }))

请注意，不可变的List 不是Seq，而是首选，因此在这里使用。如有必要，将其更改为 Seq。

测试foo:

implicit val ec = ExecutionContext.global

def fab(s: String): Future[Int] = Future{ 10 / s.length }

val ls = List("abcd", "", "xx", "")

foo(ls)(fab)
// res1: Future[List[Either[Throwable, Int]]] = Future(Success(List(
//   Right(2),
//   Left(java.lang.ArithmeticException: / by zero),
//   Right(5),
//   Left(java.lang.ArithmeticException: / by zero)
// )))

【讨论】：

非常感谢。这可能就是我正在寻找的。我只会使用Try i/o Either 并将返回类型更改为Future[List[Try[B]]]。

【解决方案2】：

我有一个 ZIO 的解决方案。

我添加了这个伪函数：

  def fab(implicit ec: ExecutionContext): Int => Future[String] = i => Future(
    if (i % 3 == 0)
      throw new IllegalArgumentException(s"bad $i")
    else
      s"$i"
  )

现在我们创建一个 Int 的 Stream 并为它们每个运行 fab

  val stream =
    ZStream.fromIterable(Seq(1, 2, 3, 4, 5))
      .map(in => Task.fromFuture(implicit ec => fab(ec)(in)))

  val sink = Sink.collectAll[Task[String]]

现在我们收集成功和失败：

  val collect: ZIO[zio.ZEnv, Throwable, (List[String], List[Throwable])] = for {
    strs <- stream.run(sink)
    successes <- Task.collectAllSuccesses(strs)
    failures <- ZIO.collectAllSuccesses(strs.map(_.flip))
  } yield (successes, failures)

运行和打印这个：

  new DefaultRuntime {}
    .unsafeRun(
      collect
        .tapError { ex => zio.console.putStrLn(s"There was an exception: ${ex.getMessage}") }
        .tap { case (successes, failures) => zio.console.putStrLn(s"($successes, $failures)") }
        .fold(_ => -1, _ => 0)
    )

打印我们：

(List(1, 2, 4, 5), List(java.lang.IllegalArgumentException: bad 3))

如果您需要更多解释，请告诉我 - 如果 ZIO 是一个选项。

【讨论】：

谢谢，但我想避免使用外部库，公司。 zio、cats等