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;
import java.util.concurrent.CountDownLatch; // 用于在main方法中等待所有异步任务完成

public class AsyncListProcessor {

    private final ExecutorService executor = Executors.newFixedThreadPool(3); // 示例用线程池，限制并发数

    // 模拟一个返回Future的耗时操作
    private Future<String> 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<String> processItemAsUni(String item) {
        return Uni.createFrom().future(processFuture(item));
    }

    public void processListReactive(List<String> items, CountDownLatch latch) {
        System.out.println("\n--- 启动非阻塞流式处理 ---");
        Uni.createFrom()
            .item(items)
            // 将 Uni<List<String>> 转换为 Multi<String>
            .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<String> 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(); // 关闭线程池
    }
}

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<String> 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<String> processItemAsUni(String item) {
        return Uni.createFrom().future(processFuture(item));
    }

    public List<String> processListBlocking(List<String> items) {
        System.out.println("\n--- 启动阻塞式处理 ---");
        List<String> results = Uni.createFrom()
            .item(items)
            .onItem().transformToMulti(Multi.createFrom()::iterable)
            .onItem().transformToUniAndMerge(this::processItemAsUni)
            .collect().asList() // 收集所有结果到一个 Uni<List<String>>
            .await().indefinitely(); // 阻塞当前线程直到所有结果收集完毕

        System.out.println("--- 阻塞式处理完成 ---");
        return results;
    }

    public static void main(String[] args) {
        AsyncListProcessorBlocking processor = new AsyncListProcessorBlocking();
        List<String> data = List.of("alpha", "beta", "gamma", "delta", "epsilon");

        List<String> processedResults = processor.processListBlocking(data);
        System.out.println("所有处理结果 (阻塞式): " + processedResults);

        processor.executor.shutdown(); // 关闭线程池
    }
}