确保 Spring Webflux 中的方法执行

Question

override fun uploadFileAndNotifyCadmium(
        validatedPages: List<KycFile>,
        documentId: UUID,
    ): Mono<Unit> {

        validatedPages.forEach { it ->
            s3Service.uploadToToxicBucket(
                documentId, it.fileName })
                .then()
        }

        return cadmiumClient.notifyCadmium(documentId) // API call to other microservice
    }

我有这个方法，我试图调用一个方法 (s3Service.uploadToToxicBucket())，它在迭代列表时返回 Unit 的 Mono。我想确保为每个列表元素调用此方法。因此，我将 then() 添加到生成的 Mono 中。此外，我希望方法 cadmiumClient.notifyCadmium() 仅在所有文件上传后才执行。这是这样做的正确方法还是我可以使用其他运算符。 cadmiumClient.notifyCadmium() 也返回 Unit 的 Mono。

当我打电话给uploadFileAndNotifyCadmium()时，我也在打电话给uploadFileAndNotifyCadmium().then()。

我使用 then() 是因为执行是惰性的，我不想调用 subscribe() 来确保这些方法的执行。此外，在控制器中调用了方法 uploadFileAndNotifyCadmium()，因此它由 Spring Webflux 自动订阅。我对 then() 用法的理解可能是错误的。

方法2：我也想过这样做：

override fun uploadFileAndNotifyCadmium(
        validatedPages: List<KycFile>,
        documentId: UUID,
    ): Mono<Unit> {

        val list = mutableListOf<Mono<Unit>>()

        validatedPages.forEach { it ->
            val result =s3Service.uploadToToxicBucket(
                documentId, it.fileName })

        list.add(result)

        }

        //Zip all the elements in list together into some variable zippedResult

        return zippedResult.then(cadmiumClient.notifyCadmium(documentId)) // API call to other microservice
    }

对于这种方法，当列表的大小以前不知道时，我找不到任何运算符将所有元素压缩在一起。

Answer 1

我试图理解你想要做什么，因为你的解释很不清楚。

我不会在 Kotlin 中编码，但如果我做对了，你想要做的是：

• 包含 n 个对象的列表
• 对列表中的每个对象执行某种类型的副作用
• 收集所有副作用的某种形式的结果
• 返回此结果

这就是我在普通反应式 java 中的做法

    return Flux.fromIterable(List.of("one", "two", "three"))
            .flatMap( // do some side effect )
            .collect(Collectors.toList());

• 将列表转换为Flux<T>
• 在flatMap 中为每个发出的项目执行副作用
• 将所有结果收集为一个列表
• 返回Mono<List<T>>