结合 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一个流，forEach 在列表的情况下）为了清楚起见。

嵌套流也有点复杂，因为您可以直接将列表修改为列表。

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

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

并行版本，如果 API 调用返回参数。 (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)
            }
}