- >这类结构时，一个常见需求是将列表中的每个元素独立地进行异步操作。例如，从数据库批量查询得到一个ID列表，然后需要为每个ID调用一个外部服务。直接对Uni
-
Uni
- > 转换为 Multi
- 元素异步转换: 对每个Multi中的元素，使用onItem().transformToUniAndMerge(item -> Uni.createFrom().future(processFuture(item)))将其转换为一个代表异步操作的Uni。transformToUniAndMerge会自动处理这些Uni的并发执行和结果合并。
- 结果处理与流终止: transformToUniAndMerge会返回一个新的Multi，其元素是所有异步操作的结果。可以通过subscribe()消费这些结果。为了确保所有异步操作完成，特别是在非Web服务器等环境中，需要额外的机制来保持主线程运行。
-
Uni
- > 转换为 Multi
- 元素异步转换: 同方法一，使用onItem().transformToUniAndMerge()。
-
收集并等待: 在transformToUniAndMerge返回的Multi上调用collect().asList()将其所有元素收集到一个Uni
- >中，然后使用await().indefinitely()阻塞当前线程，直到该Uni完成。
- >进行map操作通常会将整个列表作为一个整体处理，而无法实现对列表内每个元素的并发异步处理。本文将深入探讨Mutiny提供的强大工具，帮助开发者优雅地实现这一目标，避免常见陷阱。
核心问题剖析
当面对一个Uni
要正确地将Uni>中的每个元素转换为一个独立的异步Uni
方法一：利用Multi进行非阻塞流式处理
这种方法是Mutiny推荐的、更符合响应式编程范式的处理方式。它通过将包含列表的Uni转换为一个Multi流，然后对流中的每个元素进行异步转换和合并，实现并发处理。
原理介绍
Mutiny的Multi类型非常适合处理元素流。通过以下步骤，我们可以将Uni>转换为Multi
代码示例
以下示例演示了如何使用线程池模拟异步操作，并结合Mutiny的Multi进行非阻塞流式处理。
import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.Uni;
import java.time.Duration;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.CountDownLatch; // 用于在main方法中等待所有异步任务完成
public class AsyncListProcessor {
private final ExecutorService executor = Executors.newFixedThreadPool(3); // 示例用线程池，限制并发数
// 模拟一个返回Future的耗时操作
private Future processFuture(String s) {
return executor.submit(() -> {
System.out.println("开始处理 (Future): " + s + " on thread " + Thread.currentThread().getName());
try {
Thread.sleep(new Random().nextInt(3000) + 1000); // 模拟1-4秒延迟
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("处理中断", e);
}
System.out.println("结束处理 (Future): " + s + " on thread " + Thread.currentThread().getName());
return s.toUpperCase(); // 假设处理后返回大写
});
}
// 将Future封装成Uni
private Uni processItemAsUni(String item) {
return Uni.createFrom().future(processFuture(item));
}
public void processListReactive(List items, CountDownLatch latch) {
System.out.println("\n--- 启动非阻塞流式处理 ---");
Uni.createFrom()
.item(items)
// 将 Uni> 转换为 Multi
.onItem().transformToMulti(Multi.createFrom()::iterable)
// 对 Multi 中的每个元素进行异步处理，并合并结果
.onItem().transformToUniAndMerge(this::processItemAsUni)
// 订阅并打印每个完成的结果
.subscribe()
.with(
s -> System.out.println("接收到结果 (Reactive): " + s),
failure -> System.err.println("处理失败: " + failure.getMessage()),
() -> {
System.out.println("所有非阻塞流式处理完成.");
latch.countDown(); // 通知主线程所有任务已完成
}
);
}
public static void main(String[] args) throws InterruptedException {
AsyncListProcessor processor = new AsyncListProcessor();
List data = List.of("apple", "banana", "cherry", "date", "elderberry");
// 使用CountDownLatch等待所有异步任务完成
CountDownLatch latch = new CountDownLatch(1);
processor.processListReactive(data, latch);
// 等待所有异步任务完成
latch.await();
System.out.println("主线程继续执行，所有异步任务已完成或失败。");
processor.executor.shutdown(); // 关闭线程池
}
} 注意事项
这种方式是非阻塞的，非常适合构建响应式应用程序。它允许任务并发执行，且不会阻塞主调用线程。在非Web服务器等环境中（如简单的main方法），为了确保程序不会在异步操作完成前退出，需要使用CountDownLatch、await()或其他同步机制来等待所有任务完成。在基于Mutiny的框架（如Quarkus）中，这些通常由框架的调度器和生命周期管理。
方法二：收集并等待所有结果 (阻塞式)
如果你的需求是等待所有异步操作完成，并将它们的结果收集到一个列表中，然后才能继续执行后续逻辑，那么可以使用阻塞式的方法。
原理介绍
这种方法同样利用Multi进行元素的异步转换，但在最后阶段，它会阻塞当前线程，直到所有异步操作完成并将结果聚合到一个列表中。
代码示例
import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.Uni;
import java.time.Duration;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
public class AsyncListProcessorBlocking {
private final ExecutorService executor = Executors.newFixedThreadPool(3); // 示例用线程池
// 模拟一个返回Future的耗时操作 (同上)
private Future processFuture(String s) {
return executor.submit(() -> {
System.out.println("开始处理 (Future): " + s + " on thread " + Thread.currentThread().getName());
try {
Thread.sleep(new Random().nextInt(3000) + 1000); // 模拟1-4秒延迟
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("处理中断", e);
}
System.out.println("结束处理 (Future): " + s + " on thread " + Thread.currentThread().getName());
return s.toUpperCase();
});
}
// 将Future封装成Uni
private Uni processItemAsUni(String item) {
return Uni.createFrom().future(processFuture(item));
}
public List processListBlocking(List items) {
System.out.println("\n--- 启动阻塞式处理 ---");
List results = Uni.createFrom()
.item(items)
.onItem().transformToMulti(Multi.createFrom()::iterable)
.onItem().transformToUniAndMerge(this::processItemAsUni)
.collect().asList() // 收集所有结果到一个 Uni>
.await().indefinitely(); // 阻塞当前线程直到所有结果收集完毕
System.out.println("--- 阻塞式处理完成 ---");
return results;
}
public static void main(String[] args) {
AsyncListProcessorBlocking processor = new AsyncListProcessorBlocking();
List data = List.of("alpha", "beta", "gamma", "delta", "epsilon");
List processedResults = processor.processListBlocking(data);
System.out.println("所有处理结果 (阻塞式): " + processedResults);
processor.executor.shutdown(); // 关闭线程池
}
} 注意事项
await().indefinitely()会阻塞调用线程。虽然它能确保所有异步操作完成，但在响应式系统中应谨慎使用，因为它可能导致线程阻塞，降低系统的并发能力。它更适用于启动时的数据加载、测试场景或需要等待所有结果才能继续的特定批处理任务。在Web应用中，应避免在处理请求的线程中使用await()，以防阻塞请求处理。
总结与最佳实践
Mutiny提供了灵活且强大的机制来处理Uni>中的元素异步操作。选择哪种
