Mutiny响应式编程：异步处理Uni中元素的两种方法

Uni<List<String>> listUni = Uni.createFrom().item(List.of("a", "b", "c"));

listUni.map(strings -> strings.stream().map(this::processItem).toList()) // 将List<String>映射为List<Uni<Void>>
       .flatMap(unis -> Uni.join().all(unis).andCollectFailures()) // 合并所有Uni
       .subscribe()
       .with(System.out::println);

// 模拟异步处理的方法
private Uni<Void> processItem(String item) {
    // 假设这里是一个耗时的异步操作
    return Uni.createFrom().item(item)
               .onItem().delayIt().by(Duration.ofSeconds(1))
               .invoke(() -> System.out.println("Processing item: " + item))
               .replaceWithVoid();
}

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 {

    private final ExecutorService executor = Executors.newFixedThreadPool(3); // 用于模拟耗时操作的线程池

    // 模拟一个耗时的异步处理方法
    private Uni<String> processItemAsync(String item) {
        Random random = new Random();
        int duration = (random.nextInt(5) + 1) * 1000; // 随机延迟1-5秒
        System.out.println("Starting processing for '" + item + "' with duration " + duration + "ms on thread: " + Thread.currentThread().getName());

        // 使用Uni.createFrom().future来集成外部异步任务
        return Uni.createFrom().future(() -> executor.submit(() -> {
            try {
                Thread.sleep(duration);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            System.out.println("Finished processing for '" + item + "' on thread: " + Thread.currentThread().getName());
            return item.toUpperCase(); // 返回处理后的结果
        }));
    }

    @Test
    public void testAsyncProcessingWithVertxUnit(VertxTestContext testContext) {
        // 创建一个Uni<List<String>>
        Uni.createFrom().item(List.of("a", "b", "c", "d"))
                // 将Uni<List<String>>转换为Multi<String>，以便逐个处理元素
                .onItem().transformToMulti(Multi.createFrom()::iterable)
                // 对Multi中的每个元素应用异步转换，并合并结果
                .onItem().transformToUniAndMerge(this::processItemAsync)
                // 订阅并处理每个完成的项
                .onItem().invoke(s -> System.out.println("Received processed item: " + s))
                // 当流终止时（所有元素处理完毕或发生错误），通知测试上下文
                .onTermination().invoke((throwable, isCancelled) -> {
                    if (throwable != null) {
                        testContext.fail(throwable); // 如果有错误，测试失败
                    } else {
                        System.out.println("All items processed successfully.");
                        testContext.completeNow(); // 所有任务完成，测试成功
                    }
                    executor.shutdown(); // 关闭线程池
                })
                .subscribe()
                .with(
                        item -> System.out.println("Printing final result: " + item), // 实际的订阅者处理
                        testContext::fail // 错误处理
                );
    }
}

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;

public class AsyncListProcessingBlocking {

    private static final ExecutorService executor = Executors.newFixedThreadPool(3); // 用于模拟耗时操作的线程池

    // 模拟一个耗时的异步处理方法
    private Uni<String> processItemAsync(String item) {
        Random random = new Random();
        int duration = (random.nextInt(5) + 1) * 1000; // 随机延迟1-5秒
        System.out.println("Starting processing for '" + item + "' with duration " + duration + "ms on thread: " + Thread.currentThread().getName());

        return Uni.createFrom().future(() -> executor.submit(() -> {
            try {
                Thread.sleep(duration);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            System.out.println("Finished processing for '" + item + "' on thread: " + Thread.currentThread().getName());
            return item.toUpperCase();
        }));
    }

    public static void main(String[] args) {
        AsyncListProcessingBlocking processor = new AsyncListProcessingBlocking();

        System.out.println("Starting main processing...");

        List<String> results = Uni.createFrom().item(List.of("x", "y", "z", "w"))
                // 将Uni<List<String>>转换为Multi<String>
                .onItem().transformToMulti(Multi.createFrom()::iterable)
                // 对Multi中的每个元素应用异步转换，并合并结果
                .onItem().transformToUniAndMerge(processor::processItemAsync)
                // 打印每个完成的项（可选）
                .onItem().invoke(s -> System.out.println("Intermediate result received: " + s))
                // 收集所有结果到一个列表中，并阻塞当前线程直到所有结果可用
                .collect().asList()
                .await().indefinitely(); // 阻塞等待，直到所有结果收集完毕

        System.out.println("All items processed. Final collected results: " + results);
        executor.shutdown(); // 关闭线程池
        System.out.println("Main processing finished.");
    }
}