Mutiny异步流处理：高效并发处理Uni中的元素

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<String> 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<List<String>>
                    // 1. 将Uni<List<String>>转换为Multi<String>
                    .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测试结束
                    );
        });
    }
}

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<String> 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<String> results = Uni.createFrom()
                .item(List.of("a", "b", "c")) // 初始的Uni<List<String>>
                // 1. 将Uni<List<String>>转换为Multi<String>
                .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<List<String>>完成并返回结果
                .await().indefinitely();

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