合并 kotlin 流结果

Question

我在想是否有一种干净的方法可以在 Kotlin 中启动一系列流程，然后在解决这些问题后，根据它们是否成功执行进一步的操作

例如，我需要从数据库中读取所有整数（将它们返回到一个流中），检查它们相对于外部 API 是偶数还是奇数（也返回一个流），然后删除数据库中的奇数

在代码中应该是这样的

fun findEven() {
   db.readIntegers()
      .map { listOfInt ->
          listOfInt.asFlow()
             .flatMapMerge { singleInt ->
                httpClient.apiCallToCheckForOddity(singleInt)
                   .catch {
                      // API failure when number is even
                   }
                   .map {
                      // API success when number is odd
                      db.remove(singleInt).collect()
                   }
             }.collect()
      }.collect()
}

但是我在这段代码中看到的问题是访问数据库并行删除条目，我认为更好的解决方案是运行所有 API 调用并在某处收集所有失败了，所有的都成功了，所以能够在数据库中只进行一次批量插入，而不是让多个协程自己做这件事

Answer 1

在我看来，在 map、filter 等中产生副作用是一种反模式。从数据库中删除项目等副作用应该是一个单独的步骤 ( collect 在 Flow 的情况下，forEach 在 List 的情况下）为了清楚起见。

嵌套流程也比较复杂，因为您可以直接将列表修改为 List。

我认为你可以这样做，假设 API 一次只能检查一项。

suspend fun findEven() {
    db.readIntegers()
            .map { listOfInt ->
                listOfInt.filter { singleInt ->
                    runCatching {
                        httpClient.apiCallToCheckForOddity(singleInt)
                    }.isSuccess
                }
            }
            .collect { listOfOddInt ->
                db.removeAll(listOfOddInt)
            }
}

并行版本，如果API调用returns参数。 (By the way, Kotlin APIs should not throw exceptions on non-programmer errors).

suspend fun findEven() {
    db.readIntegers()
            .map { listOfInt ->
                coroutineScope {
                    listOfInt.map { singleInt ->
                        async {
                            runCatching {
                                httpClient.apiCallToCheckForOddity(singleInt)
                            }
                        }
                    }.awaitAll()
                            .mapNotNull(Result<Int>::getOrNull)
                }
            }
            .collect { listOfOddInt ->
                db.removeAll(listOfOddInt)
            }
}

合并 kotlin 流结果

Combining kotlin flow results

concurrency

kotlin

kotlin-coroutines

kotlin-flow