17370845950

理解Uni中的异步处理挑战

在quarkus或类似的响应式编程环境中，uni>表示一个最终会发出一个list的异步操作。当我们需要对这个列表中的每个元素执行另一个异步操作时，常见的挑战是如何确保所有这些独立的异步操作都能被正确地触发、执行并等待其完成，而不会因为主线程的提前退出而导致部分操作未能完成。

初始尝试可能涉及将Uni>通过map操作转换为List>，然后使用Uni.join().all(unis).andCollectFailures()来合并这些Uni。然而，如果后续直接使用.subscribe().with(System.out::println)，由于subscribe是非阻塞的，程序可能会在所有异步任务完成之前就退出，从而给人一种只有第一个元素被处理的错觉。核心问题在于如何管理这些异步操作的生命周期，确保在所有任务真正完成之前，程序不会终止。

解决方案一：利用Multi进行流式处理与异步生命周期管理

为了高效且可靠地处理Uni中的每个元素，并确保所有异步操作都能在主程序退出前完成，我们可以将Uni转换为Multi，从而将列表中的每个元素作为独立的流事件来处理。结合Vert.x Unit等测试框架的异步上下文，可以优雅地管理异步流的生命周期。

核心思路

1. Uni转换为Multi：Uni.onItem().transformToMulti(Multi.createFrom()::iterable)是关键一步，它将一个包含列表的Uni转换为一个可以逐个发出列表元素的Multi。
2. 并发处理每个元素：onItem().transformToUniAndMerge(s -> ...)允许我们为Multi中的每个元素创建一个新的Uni（代表该元素的异步处理），然后将这些独立的Uni的发射合并回一个单一的Multi。transformToUniAndMerge的特性是它会并发地订阅和处理这些内部的Uni。
3. 异步生命周期管理：在测试环境中，Vert.x Unit的TestContext和Async对象提供了一种机制来明确地通知测试框架何时异步操作完成。onTermination().invoke()可以在流终止时执行回调，用于标记异步任务的完成状态。

示例代码 (结合Vert.x Unit)

import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.Uni;
import io.vertx.ext.unit.Async;
import io.vertx.ext.unit.TestContext;
import io.vertx.ext.unit.junit.VertxUnitRunner;
import org.junit.Test;
import org.junit.runner.RunWith;

import java.time.Duration;
import java.util.List;
import java.util.Random;

@RunWith(VertxUnitRunner.class)
public class AsyncListProcessingExample {

    @Test
    public void testAsyncListProcessing(TestContext context) {
        Random random = new Random();
        // 用于在流终止时通知测试框架
        Async async = context.async();

        Uni.createFrom()
                .item(List.of("a", "b", "c")) // 原始的Uni>
                // 将 Uni> 转换为 Multi，每个字符串作为独立的流事件
                .onItem().transformToMulti(Multi.createFrom()::iterable)
                // 对 Multi 中的每个字符串元素进行异步处理
                .onItem().transformToUniAndMerge(s ->
                        // 为每个元素创建一个 Uni，模拟异步操作（这里是随机延迟）
                        Uni.createFrom().item(s)
                                .onItem().delayIt().by(Duration.ofMillis((random.nextInt(5) + 1) * 1000))
                )
                // 在流终止时执行回调，无论成功或失败
                .onTermination().invoke((throwable, aBoolean) -> {
                    if (throwable != null) {
                        context.fail(throwable); // 如果有异常，测试失败
                    } else {
                        async.complete(); // 所有异步操作完成，通知测试通过
                    }
                })
                // 订阅 Multi，处理每个完成的元素
                .subscribe()
                .with(s -> System.out.println("Printing: " + s));
    }
}

注意事项

• 测试上下文：VertxUnitRunner和TestContext是测试框架提供的，用于确保异步测试的正确执行。在生产代码中，你可能需要根据实际应用场景（如Web请求的生命周期、后台服务的启动/停止）来管理异步操作的完成。
• transformToUniAndMerge：这个操作符是实现并发处理的关键。它会为每个流元素创建一个Uni，并并发地订阅这些Uni，然后将它们的结果按完成顺序合并到下游。

解决方案二：阻塞式收集所有结果

在某些场景下，我们可能需要等待所有异步操作完成后，才继续执行后续的同步代码，或者需要将所有异步处理的结果收集到一个列表中。Mutiny提供了阻塞式等待和收集结果的机制。

核心思路

1. Uni转换为Multi：与解决方案一相同，首先将Uni转换为Multi。
2. 并发处理并收集：同样使用onItem().transformToUniAndMerge()进行并发处理。
3. 阻塞等待并收集结果：collect().asList().await().indefinitely()是一个强大的组合。collect().asList()会将Multi发出的所有元素收集到一个List中，并返回一个Uni。接着，.await().indefinitely()会阻塞当前线程，直到这个Uni发出其结果（即收集到的列表），或者发生错误。

示例代码 (阻塞式收集)

import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.Uni;

import java.time.Duration;
import java.util.List;
import java.util.Random;

public class BlockingListProcessingExample {

    public static void main(String[] args) {
        Random random = new Random();

        List results = Uni.createFrom()
                .item(List.of("a", "b", "c")) // 原始的Uni>
                // 将 Uni> 转换为 Multi
                .onItem().transformToMulti(Multi.createFrom()::iterable)
                // 对 Multi 中的每个字符串元素进行异步处理
                .onItem().transformToUniAndMerge(s -> {
                    final int duration = (random.nextInt(5) + 1) * 1000;
                    // 为每个元素创建一个 Uni，模拟异步操作
                    return Uni.createFrom().item(s)
                            .onItem().delayIt().by(Duration.ofMillis(duration))
                            .invoke(() -> System.out.println("Letter: " + s + ", duration in ms: " + duration));
                })
                // 订阅 Multi，处理每个完成的元素
                .onItem().invoke(s -> System.out.println("Printing: " + s))
                // 收集所有结果到一个列表中
                .collect().asList()
                // 阻塞当前线程，直到所有结果都被收集完毕
                .await().indefinitely();

        System.out.println("All results collected: " + results);
    }
}

注意事项

• 阻塞操作：.await().indefinitely()会阻塞当前线程。在响应式应用中，应尽量避免在I/O线程或事件循环线程中执行阻塞操作，因为它会降低系统的吞吐量。此方法更适用于主方法、命令行工具或需要在特定点同步等待结果的场景。
• 错误处理：如果任何一个内部的Uni失败，整个流会终止，并且await()方法会抛出异常。你可以使用onFailure().recoverWith(...)等操作符来处理错误。

总结

在Mutiny中异步处理Uni中的元素，关键在于将Uni有效地转换为Multi，以便对每个元素进行流式处理。

• 对于需要非阻塞、响应式处理且在测试环境中管理异步操作生命周期的场景，onItem().transformToMulti(Multi.createFrom()::iterable)结合onItem().transformToUniAndMerge() 是理想选择，并通过测试上下文（如Vert.x Unit的Async）来确保所有异步任务的完成。
• 当需要在某个点同步等待所有异步操作完成并收集其结果时，collect().asList().await().indefinitely() 提供了一种方便的阻塞式解决方案，但需注意其对性能和响应性的潜在影响。

理解这两种模式及其适用场景，能够帮助开发者更灵活、高效地构建基于Mutiny的异步应用。