<img src="https://img.php.cn/upload/article/001/246/273/175894634155686.jpg" alt="异步处理Uni中的元素：从问题到解决方案”>中的元素：从问题到解决方案” />
本文深入探讨了如何在响应式编程框架中异步处理Uni类型中的每个元素。针对常见的异步操作生命周期管理挑战，文章详细介绍了两种核心策略：一是通过将Uni转换为Multi并利用transformToUniAndMerge实现并发处理，并结合测试上下文确保异步流的完整性；二是通过collect().asList().await().indefinitely()阻塞式等待所有异步操作完成并收集结果。文章旨在提供清晰的指导和示例，帮助开发者有效管理和执行列表元素的并发异步处理。
理解Uni中的异步处理挑战
在quarkus或类似的响应式编程环境中，uni<list>表示一个最终会发出一个list的异步操作。当我们需要对这个列表中的每个元素执行另一个异步操作时，常见的挑战是如何确保所有这些独立的异步操作都能被正确地触发、执行并等待其完成，而不会因为主线程的提前退出而导致部分操作未能完成。
初始尝试可能涉及将Uni<List>通过map操作转换为List<Uni>，然后使用Uni.join().all(unis).andCollectFailures()来合并这些Uni。然而，如果后续直接使用.subscribe().with(System.out::println)，由于subscribe是非阻塞的，程序可能会在所有异步任务完成之前就退出，从而给人一种只有第一个元素被处理的错觉。核心问题在于如何管理这些异步操作的生命周期，确保在所有任务真正完成之前，程序不会终止。
解决方案一：利用Multi进行流式处理与异步生命周期管理
为了高效且可靠地处理Uni中的每个元素，并确保所有异步操作都能在主程序退出前完成，我们可以将Uni转换为Multi，从而将列表中的每个元素作为独立的流事件来处理。结合Vert.x Unit等测试框架的异步上下文，可以优雅地管理异步流的生命周期。
核心思路
Uni转换为Multi：Uni.onItem().transformToMulti(Multi.createFrom()::iterable)是关键一步，它将一个包含列表的Uni转换为一个可以逐个发出列表元素的Multi。并发处理每个元素：onItem().transformToUniAndMerge(s -> …)允许我们为Multi中的每个元素创建一个新的Uni（代表该元素的异步处理），然后将这些独立的Uni的发射合并回一个单一的Multi。transformToUniAndMerge的特性是它会并发地订阅和处理这些内部的Uni。异步生命周期管理：在测试环境中，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<List> // 将 Uni<List> 转换为 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提供了阻塞式等待和收集结果的机制。
核心思路
Uni转换为Multi：与解决方案一相同，首先将Uni转换为Multi。并发处理并收集：同样使用onItem().transformToUniAndMerge()进行并发处理。阻塞等待并收集结果：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<List> // 将 Uni<List> 转换为 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的异步应用。
以上就是异步处理Uni中的元素：从问题到解决方案的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/53064.html
微信扫一扫 
支付宝扫一扫 
