使用 Arrow-kt 和 Kotlin 处理异步结果

Question

我有两个对外部系统的异步函数调用返回 Either 并且需要合并它们的结果。作为 Arrow-Kt 中函数式编程的初学者，我想知道哪种方法是完成此任务的最佳方法。下面是我目前正在使用的代码。它当然有效，但并不是真正“感觉”是最直接的。我正在寻找一种更“实用”的风格来获得结果。注意：成功列表结果的前期使用是必要的。

suspend fun getAs(): Either<Exception, List<A>> = TODO()
suspend fun getBs(): Either<Exception, List<B>> = TODO()
suspend fun doSomethingWithA(listA: List<A>): Unit = TODO()

launch {
    val deferredA = async { getAs() }
    val deferredB = async { getBs() }

    either<Exception, List<A>> {
        val listOfAs = deferredA.await()
            .bimap(leftOperation = { e ->
                println("special message on error for A")
                e
            }, rightOperation = { listA ->
                doSomethingWithA(listA)
                listA
            })
            .bind()
        val listOfBs = deferredB.await().bind()

        listOfAs.filter { it.someId !in listOfBs.map { it.someProperty } }
    }
    .map { /* handle result */ }
    .handleError { /* handle error */ }

}

另一种选择是像这样使用 map{} 函数

launch {
    val deferredA = async { getAs() }
    val deferredB = async { getBs() }

    deferredA.await()
        .bimap(leftOperation = { e ->
            println("special message on error for A")
            e
        }, rightOperation = { listA ->
            doSomethingWithA(listA)
            deferredB.await().map { listB ->
                listA.filter { a -> a.someId !in listB.map { it.someProperty } }
            }
        })
        .map { /* handle result */ }
        .handleError { /* handle error */ }
}

Answer 1

最简单的方法是将 either { } 与 parZip 合并。 either { } 允许您从 Either<E, A> 中提取 A，而 parZip 是运行 suspend 函数并行的效用函数。

suspend fun getAs(): Either<Exception, List<A>> = TODO()
suspend fun getBs(): Either<Exception, List<B>> = TODO()
suspend fun doSomethingWithA(listA: List<A>): Unit = TODO()

either {
  val list = parZip(
    {
       getAs()
         .mapLeft { e -> println("special message on error for A"); e }
         .bind()
    },
    { getBs().bind() },
    { aas, bbs ->
      aas.filter { a -> a.someId !in bbs.map { it.someProperty }
    }
  )

  /* Work with list and return value to `either { } */
}.handleError { /* handle error */ }

这里 bind() 从 Either<E, A> 中提取 A。我们在 parZip 中执行此操作，这样每当遇到 Left 时，它就会使 either { } 块短路，并且通过这样做，它还会取消 [=] 中仍然存在的运行任务12=].

这样如果 getAs() returns 立即与 Left 然后它成为 either { } 的输出值并且 getBs() 被取消。

Answer 2

我正要 post 一个非常相似的答案。请注意 getAs 和 getBs 并不是真正的顺序 ，因为 getBs 不需要执行 getAs 的结果。他们只是碰巧需要最终合并结果。换句话说：我们可以并行化

除了 Simon 的建议之外，我还会做一些额外的事情。（在此示例中，我将用 NetworkUser 和 DbUser 替换 A 和 B，以尝试赋予它一些语义，否则过滤器上的那些“id”属性将不起作用。

捕获错误并将它们映射到每个有效函数上的强类型域错误。

这将有助于减轻程序其余部分的负担，并在其之上提供更安全的域错误层次结构，我们可以在需要时对其进行详尽评估。

suspend fun <A> getUsersFromNetwork(): Either<DomainError, List<NetworkUser>> =
 Either.catch { fetchUsers() }
   .mapLeft { exception ->
     println("special message on error for A")
     exception.toDomain()
   }

创建 doSomething 函数 return 以防它也可能失败。

这是您所说的在初始 get 后立即需要的函数，这意味着 flatMap 或 bind（它们是等效的）。如果我们将它提升到 Either 中，这将确保错误短路按预期发生，因此此操作永远不会运行在初始未成功的情况下。

我建议这样做是因为我怀疑你在这里的这个操作也是你代码中第一个操作的结果，可能是将第一个操作的结果存储在本地缓存或其他类型的缓存中效果就是消耗那个结果。

suspend fun doSomethingWithNetworkUsers(listA: List<NetworkUser>): Either<DomainError, Unit> = TODO()

因此我们将依赖组合函数的函数如下所示：

suspend fun getUsersFromNetwork(): Either<DomainError, List<NetworkUser>> = TODO()
suspend fun getUsersFromDb(): Either<DomainError, List<DbUser>> = TODO()
suspend fun doSomethingWithNetworkUsers(listA: List<NetworkUser>): Either<DomainError, Unit> = TODO()

和程序：

fun CoroutineScope.program() {
  launch {
    either {
      parZip(
        {
          val networkUsers = getUsersFromNetwork().bind()
          doSomethingWithNetworkUsers(networkUsers).bind()
          networkUsers
        },
        { getUsersFromDb().bind() }
      ) { networkUsers, dbUsers ->
        networkUsers.filter { networkUser ->
          networkUser.id !in dbUsers.map { dbUser -> dbUser.id }
        }
      }
    }
    .map { /* do something with the overall result */ }
    .handleError { /* can recover from errors here */ }
    // Alternatively:
    // .fold(ifLeft = {}, ifRight = {}) for handling both sides.
  }
}

通过将第一个操作作为组合操作首先进行绑定，就像从上面的代码中提取的以下代码片段一样，我们确保这两个操作都在 parZip lambda 组合结果发生之前完成。

val networkUsers = getUsersFromNetwork().bind()
doSomethingWithNetworkUsers(networkUsers).bind()
networkUsers

使用 Arrow-kt 和 Kotlin 处理异步结果

Processing async results with Arrow-kt and Kotlin

either

async-await

kotlin

arrow-kt