如何将上下文传播到 Quarkus 中的并行流操作？

Question

我有一系列简单的链式操作，可以使用在 Quarkus 服务中运行的 Panache 存储库检索和保存一些数据。当这些操作并行时，ContextNotActiveException会抛出 a 。在删除并行化的情况下，代码将按预期工作。

这段代码的工作原理：

    dataRepository.get()
        .map { convert(it) }
        .forEach { perist(it) }

这段代码没有：

    dataRepository.get()
        .parallelStream()
        .map { convert(it) }
        .forEach { perist(it) }

Quarkus 文档非常有限，仅解决了 mutiny 或 RX 的使用。

我怎样才能传播上下文parallelStream()才能起作用？

Answer 1

不幸的是，上下文传播不能很好地处理并行 Java 流，因为使流并行会自动将执行移动到 ForkJoinPool，这意味着您会丢失上下文。您需要以不同的方式处理并行性，而不需要 Java 流为您做这件事 - 您可能希望使用org.eclipse.microprofile.context.ManagedExecutor.

假设该convert方法无论出于何种原因都需要活动请求上下文，您将需要将其调用分派到托管执行器中。这将确保传播上下文。在 Java 代码中，我能想到的与您的代码非常相似的代码是：

    @Inject
    org.eclipse.microprofile.context.ManagedExecutor executor;

(...)

dataRepository.streamAll()
                .forEach(i -> {
                    executor.supplyAsync(() -> {
                        return convert(i);
                    }).thenAccept(persist(i));
                });