Quarkus OpenTelemetry上下文在并行流中自动传播方案咨询

方案1：使用OpenTelemetry Java Agent（推荐） #

这是最零侵入的方案，无需修改业务代码。OpenTelemetry Java Agent会自动增强JDK中的ForkJoinPool实现，在任务提交时自动捕获当前的OpenTelemetry上下文，并在任务执行时恢复上下文，自然解决并行流中的链路断裂问题。

只需在应用启动时添加Java Agent参数即可：

java -javaagent:path/to/opentelemetry-javaagent.jar -jar your-app.jar

方案2：自定义上下文感知的ForkJoinPool #

如果无法使用Agent，可以自定义一个能自动传播上下文的ForkJoinPool，替代默认的公共池来执行并行流任务：

步骤1：实现上下文传播的任务包装类 #

import io.opentelemetry.context.Context;
import io.opentelemetry.context.Scope;

public class ContextPropagatingRunnable implements Runnable {
    private final Runnable delegate;
    private final Context parentContext;

    public ContextPropagatingRunnable(Runnable delegate) {
        this.delegate = delegate;
        this.parentContext = Context.current();
    }

    @Override
    public void run() {
        // 执行任务前绑定父上下文，执行后自动释放
        try (Scope ignored = parentContext.makeCurrent()) {
            delegate.run();
        }
    }
}

步骤2：自定义ForkJoinPool #

import java.util.concurrent.ForkJoinPool;

public class ContextPropagatingForkJoinPool extends ForkJoinPool {
    public ContextPropagatingForkJoinPool() {
        super();
    }

    public ContextPropagatingForkJoinPool(int parallelism) {
        super(parallelism);
    }

    @Override
    public void execute(Runnable task) {
        // 提交任务前包装成带上下文的Runnable
        super.execute(new ContextPropagatingRunnable(task));
    }
}

步骤3：使用自定义池执行并行流 #

// 创建自定义池（可根据需求指定并行度）
ContextPropagatingForkJoinPool contextPool = new ContextPropagatingForkJoinPool();

// 提交并行流任务到自定义池执行
contextPool.submit(() -> sampleList.parallelStream().forEach(e -> e.process())).join();

方案3：手动包装Stream操作（不推荐） #

如果上述方案都不适用，可以手动将forEach中的逻辑包装成带上下文的执行，但这种方式需要修改业务代码，繁琐且容易遗漏：

Context parentContext = Context.current();
sampleList.parallelStream().forEach(e -> {
    try (Scope ignored = parentContext.makeCurrent()) {
        e.process();
    }
});

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