异步处理Uni中的元素：从问题到解决方案

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<String>>
                // 将 Uni<List<String>> 转换为 Multi<String>，每个字符串作为独立的流事件
                .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));
    }
}

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<String> results = Uni.createFrom()
                .item(List.of("a", "b", "c")) // 原始的Uni<List<String>>
                // 将 Uni<List<String>> 转换为 Multi<String>
                .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);
    }
}