Quarkus Multi多线程疑问：为何未按预期异步并行执行？

首先明确：Mutiny（Quarkus默认响应式框架）的Multi流默认是顺序执行且绑定到单个Vert.x EventLoop线程，这是Vert.x线程模型的设计——EventLoop线程单线程处理事件，避免线程安全问题，同时保证任务执行顺序。这就是你看到所有任务在同一个vert.x-eventloop-thread上顺序执行的原因。

为什么runSubscriptionOn(executor)没用？ #

runSubscriptionOn仅控制订阅阶段的执行线程（比如Multi的创建、订阅触发逻辑），但流中每个item的处理逻辑依然会回到EventLoop线程执行，所以无法实现item级别的并行处理。

实现并行处理的正确方式 #

要让Multi的item并行执行、快任务先完成输出，需要用以下操作符配合线程池（推荐使用Quarkus内置的Worker线程池，避免手动创建线程池带来的资源泄漏）：

1. 使用mapAsync指定并行度

mapAsync允许你为每个item的处理指定并行执行的线程池，同时设置最大并行度：

@Inject
@Named("worker")
Executor workerExecutor;

public Multi<Integer> processItems() {
    return Multi.createFrom().items(1, 2, 3, 4)
        // 设置并行度为4，最多同时处理4个item
        .mapAsync(4, item -> {
            // 模拟耗时任务，每个任务耗时随机
            try {
                Thread.sleep(new Random().nextInt(1000));
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            System.out.printf("处理item %d 线程: %s%n", item, Thread.currentThread().getName());
            return Uni.createFrom().item(item);
        });
}

这种方式下，item会在Worker线程池并行处理，处理完成的item会立即输出（乱序），如果需要保留原顺序，改用mapAsyncOrdered即可。

2. 使用flatMap结合Uni的线程切换

通过flatMap将每个item转换为一个在Worker线程执行的Uni，再合并结果：

@Inject
@Named("worker")
Executor workerExecutor;

public Multi<Integer> processItems() {
    return Multi.createFrom().items(1, 2, 3, 4)
        .flatMap(item -> Uni.createFrom().item(item)
            .runSubscriptionOn(workerExecutor)
            .invoke(() -> {
                try {
                    Thread.sleep(new Random().nextInt(1000));
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                System.out.printf("处理item %d 线程: %s%n", item, Thread.currentThread().getName());
            }));
}

这种方式同样会实现并行处理，结果按完成顺序输出。

关键注意事项 #

• Quarkus中优先使用内置的Worker线程池（通过@Inject @Named("worker")注入），不要手动创建ExecutorService，避免与Quarkus的线程管理冲突。
• EventLoop线程用于IO密集型操作，不要在EventLoop线程执行CPU密集或耗时任务，否则会阻塞事件循环，影响整个应用性能。

内容的提问来源于stack exchange，提问作者SilverLuke