17370845950

1. 问题背景与常见误区

在响应式编程中，我们经常会遇到需要处理包含多个元素的异步操作。例如，有一个uni>，我们希望对列表中的每个字符串都执行一个耗时的异步任务，并最终收集或处理所有任务的结果。

一个常见的尝试是使用map将List转换为List>，然后通过Uni.join().all(unis).andCollectFailures()来合并这些Uni。然而，这种方法可能无法达到预期的并发处理效果，或者在短生命周期的程序（如单元测试）中，由于主线程过早退出，导致异步任务未能完成就被终止，从而给人一种“只处理了第一个元素”的错觉。

问题的核心在于，Uni>本身代表的是一个单值流，其值是一个完整的列表。如果想对列表中的每个元素进行异步操作，并将其视为独立的响应式事件，就需要将这个列表“展开”成一个可以逐个处理的流。Mutiny提供了Multi类型来处理零到N个元素的流，这正是解决此类问题的关键。

2. Mutiny异步流处理核心：Uni与Multi

Mutiny是Quarkus等框架中广泛使用的响应式编程库，它提供了两种核心类型：

• Uni: 代表一个异步操作，最终会发出0个或1个元素，或者一个失败事件。
• Multi: 代表一个异步操作流，可以发出0到N个元素，或者一个失败事件，最终会发出完成事件。

要实现对Uni>中每个元素的异步并发处理，我们需要将Uni>首先转换为一个Multi，这样列表中的每个字符串就成为了Multi流中的一个独立事件。然后，我们可以对这个Multi流中的每个事件应用异步转换。

3. 解决方案一：在测试环境中优雅地处理异步流（结合Vert.x Unit）

在单元测试或需要非阻塞等待所有异步操作完成的场景中，我们可以利用Multi的特性和onTermination().invoke()回调来确保所有任务执行完毕。以下示例结合了Vert.x Unit，它提供了一个Async机制来管理异步测试的生命周期。

import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.Uni;
import io.vertx.core.Vertx;
import io.vertx.junit5.VertxExtension;
import io.vertx.junit5.VertxTestContext;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;

import java.time.Duration;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@ExtendWith(VertxExtension.class)
public class AsyncListProcessingTest {

    // 模拟一个异步操作，返回一个Uni
    private Uni processItemAsync(String item, Random random) {
        final int duration = (random.nextInt(5) + 1) * 1000; // 随机延迟1-5秒
        System.out.println("Starting process for: " + item + ", duration: " + duration + "ms");
        return Uni.createFrom().item(item)
                .onItem().delayIt().by(Duration.ofMillis(duration))
                .invoke(() -> System.out.println("Finished process for: " + item));
    }

    @Test
    public void testAsyncProcessingWithVertxUnit(VertxTestContext context) {
        Random random = new Random();
        // Vert.x Unit的Async对象，用于通知测试框架异步操作何时完成
        context.verify(() -> { // 确保在VertxTestContext的上下文中执行
            Uni.createFrom()
                    .item(List.of("a", "b", "c")) // 初始的Uni>
                    // 1. 将Uni>转换为Multi
                    .onItem().transformToMulti(Multi.createFrom()::iterable)
                    // 2. 对Multi中的每个元素应用异步转换，并将结果合并回Multi
                    .onItem().transformToUniAndMerge(s -> processItemAsync(s, random))
                    // 3. 订阅Multi流，处理每个完成的元素
                    .subscribe()
                    .with(
                            s -> System.out.println("Printing result: " + s), // 成功处理每个元素
                            context::failNow, // 任何错误导致流失败
                            context::completeNow // 流完成，通知VertxTestContext测试结束
                    );
        });
    }
}

代码解释：

1. Uni.createFrom().item(List.of("a", "b", "c")): 创建一个包含字符串列表的初始Uni。
2. .onItem().transformToMulti(Multi.createFrom()::iterable): 这是将Uni>转换为Multi的关键步骤。它将Uni发出的列表内容展开，使得列表中的每个元素都成为一个新的Multi事件。
3. .onItem().transformToUniAndMerge(s -> processItemAsync(s, random)): 这是实现并发异步处理的核心。
   • transformToUni: 对Multi中的每个元素s，都会调用processItemAsync(s, random)方法，该方法返回一个Uni。
   • andMerge: Mutiny会并发地订阅并执行这些由transformToUni创建的Uni。当任何一个Uni完成时，它的结果会被立即合并到输出的Multi流中。这意味着结果的顺序可能与原始列表的顺序不同，而是取决于哪个异步操作首先完成。
4. .subscribe().with(...): 订阅最终的Multi流。
   • 第一个Lambda表达式处理Multi中发出的每个成功结果。
   • 第二个Lambda表达式处理流中的任何错误。
   • 第三个Lambda表达式在流成功完成时被调用，这里我们使用context::completeNow来通知Vert.x Unit测试已成功完成所有异步操作。

4. 解决方案二：阻塞式等待所有异步结果

在某些场景下，例如在命令行工具或需要等待所有异步操作完成后才能继续主程序执行时，我们可以选择阻塞当前线程直到所有结果都被收集。

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

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

public class BlockingAsyncListProcessing {

    private static Uni processItemAsync(String item, Random random) {
        final int duration = (random.nextInt(5) + 1) * 1000; // 随机延迟1-5秒
        return Uni.createFrom().item(item)
                .onItem().delayIt().by(Duration.ofMillis(duration))
                .invoke(() -> System.out.println("Letter: " + item + ", duration in ms: " + duration));
    }

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

        System.out.println("Starting blocking asynchronous processing...");

        List results = Uni.createFrom()
                .item(List.of("a", "b", "c")) // 初始的Uni>
                // 1. 将Uni>转换为Multi
                .onItem().transformToMulti(Multi.createFrom()::iterable)
                // 2. 对Multi中的每个元素应用异步转换，并将结果合并回Multi
                .onItem().transformToUniAndMerge(s -> processItemAsync(s, random))
                // 3. 可选：处理每个完成的元素
                .onItem().invoke(s -> System.out.println("Printing collected item: " + s))
                // 4. 将Multi中的所有元素收集到一个列表中
                .collect().asList()
                // 5. 阻塞当前线程，直到Uni>完成并返回结果
                .await().indefinitely();

        System.out.println("All items processed. Collected results: " + results);
    }
}

代码解释：

1. 前两步与解决方案一相同：将Uni>转换为Multi，然后使用onItem().transformToUniAndMerge()并发处理每个元素。
2. .collect().asList(): 这个操作符将Multi流中所有发出的元素收集到一个List中，并最终返回一个Uni>。这个Uni会在源Multi完成时发出包含所有收集元素的列表。
3. .await().indefinitely(): 这是阻塞操作。它会阻塞当前线程，直到上游的Uni>发出其结果（即所有异步操作完成且结果被收集到列表中）。indefinitely()表示无限期等待。

5. 注意事项与最佳实践

• 非阻塞优先: 尽可能采用非阻塞的响应式模式（如解决方案一）。await()操作会阻塞当前线程，在生产环境中应谨慎使用，尤其是在I/O密集型或Web应用中，它可能导致线程饥饿和性能问题。它更适合于启动代码、测试或需要同步等待所有异步任务完成的特定场景。
• 错误处理: 在transformToUniAndMerge内部创建的Uni中，以及最终的subscribe().with()方法中，都应该有完善的错误处理逻辑。Mutiny提供了丰富的错误处理操作符，如onFailure().recoverWith()、onFailure().retry()等。
• 并发度: transformToUniAndMerge会并发处理任务，但实际的并发度可能受限于底层线程池配置、系统资源以及具体异步操作的实现。如果需要精细控制并发度，可以考虑使用transformToUniAndMerge(concurrency, ...)变体。
• 顺序保证: transformToUniAndMerge不保证结果的顺序与原始列表的顺序一致。如果需要保持顺序，可以考虑使用transformToUniAndConcatenate或在收集后手动排序。
• 资源管理: 确保异步操作中使用的任何外部资源（如数据库连接、文件句柄）都能得到妥善管理和释放。

6. 总结

通过Mutiny的Multi类型和onItem().transformToUniAndMerge()操作符，我们可以有效地将Uni>中的每个元素转换为独立的异步任务并进行并发处理。根据应用场景的不同，我们可以选择非阻塞的订阅模式（适用于响应式系统和测试）或阻塞式的await()模式（适用于需要同步等待结果的特定场景）。理解并正确运用这些Mutiny操作符是构建高效、健壮的响应式应用程序的关键。