Mutiny异步处理Uni中的元素：从列表到并发流

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 MutinyAsyncListProcessor {

    // 模拟一个异步耗时操作
    private Uni<String> processItemAsync(String item) {
        Random random = new Random();
        final int duration = (random.nextInt(5) + 1) * 1000; // 1-5秒随机延迟

        System.out.println("开始处理: " + item + ", 预计耗时: " + duration + "ms");

        return Uni.createFrom().item(item)
                   .onItem().delayIt().by(Duration.ofMillis(duration))
                   .invoke(() -> System.out.println("完成处理: " + item + ", 实际耗时: " + duration + "ms"));
    }

    public void processListItems() {
        Uni.createFrom()
                .item(List.of("a", "b", "c", "d", "e"))
                // 1. 将 Uni<List<String>> 转换为 Multi<String>
                .onItem().transformToMulti(Multi.createFrom()::iterable)
                // 2. 对 Multi 中的每个元素异步处理，并合并结果
                .onItem().transformToUniAndMerge(this::processItemAsync)
                // 3. 订阅并处理每个完成的异步结果
                .subscribe()
                .with(
                    s -> System.out.println("打印结果: " + s), // 成功处理每个元素后的回调
                    failure -> System.err.println("处理失败: " + failure.getMessage()), // 错误回调
                    () -> System.out.println("所有元素处理完毕！") // 流完成回调
                );

        // 注意：在实际应用中，如果这是主线程，需要一个机制来等待，
        // 否则程序可能在异步任务完成前退出。
        // 例如，在测试中可以使用 TestContext.async()，
        // 在实际服务中，这些通常由框架（如Quarkus）的事件循环管理。
        // 为了演示，这里添加一个简单的阻塞，但在响应式编程中应避免。
        try {
            Thread.sleep(8000); // 等待足够长时间，确保所有异步任务完成
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) {
        new MutinyAsyncListProcessor().processListItems();
    }
}

import io.smallrye.mutiny.Multi;
import io.smallrye.mutiny.Uni;
import java.time.Duration;
import java.util.List;
import java.util.Random;

public class MutinyCollectAndAwait {

    // 模拟一个异步耗时操作
    private Uni<String> processItemAsync(String item) {
        Random random = new Random();
        final int duration = (random.nextInt(5) + 1) * 1000; // 1-5秒随机延迟

        System.out.println("开始处理: " + item + ", 预计耗时: " + duration + "ms");

        return Uni.createFrom().item(item)
                   .onItem().delayIt().by(Duration.ofMillis(duration))
                   .invoke(() -> System.out.println("完成处理: " + item + ", 实际耗时: " + duration + "ms"));
    }

    public void collectAndAwaitResults() {
        List<String> results = Uni.createFrom()
                .item(List.of("x", "y", "z", "w"))
                // 1. 将 Uni<List<String>> 转换为 Multi<String>
                .onItem().transformToMulti(Multi.createFrom()::iterable)
                // 2. 对 Multi 中的每个元素异步处理，并合并结果
                .onItem().transformToUniAndMerge(this::processItemAsync)
                // 3. 收集所有异步处理的结果到一个 List<String> 中
                .collect().asList()
                // 4. 阻塞当前线程，直到所有结果都收集完毕
                .await().indefinitely(); // 或者 await().atMost(Duration.ofSeconds(10))

        System.out.println("\n所有异步处理完成，收集到的结果: " + results);
    }

    public static void main(String[] args) {
        new MutinyCollectAndAwait().collectAndAwaitResults();
    }
}