为虚拟线程任务执行器适配Micrometer链路追踪的问题咨询

原始代码 #

@PostMapping("/question")
public String question(@RequestBody List<String> messages) {
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        for (String message : messages) {
            executor.submit(() -> {
                kafkaTemplate.send("topic-loom-micrometer", message + Thread.currentThread().getName());
            });
        }
        executor.shutdown();
    }
    return "it seems everything happened correctly";
}

这段代码从REST接口接收字符串列表，通过虚拟线程并行发送消息到Kafka，提升海量消息的并行处理效率。

需求 #

基于Micrometer实现链路追踪观测，参考Micrometer官方观测文档中线程切换组件的 instrumentation 内容。

遇到的问题 #

• 文档示例使用Executors.newCachedThreadPool()，不确定是否适用于虚拟线程；
• 不清楚then(registry.getCurrentObservation()).isSameAs(parent);的来源与用法；
• ContextExecutorService.wrap(executor)已弃用，文档未说明替代方案；
• 尝试适配的代码存在编译和逻辑问题，代码如下：

@PostMapping("/question2")
public String question2(@RequestBody List<String> messages) {
    try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        Observation parent = Observation.createNotStarted("parent", registry);
        for (String message : messages) {
            parent.observe(() -> {
                ContextExecutorService.wrap(executor).submit(() -> {
                    return Observation.createNotStarted("child", registry)
                            .observe(() -> {
                                kafkaTemplate.send("topic-loom-micrometer", message + Thread.currentThread().getName());
                            });
                        }
                );
            });
        }
        executor.shutdown();
    }
}

额外疑问 #

A. 结合for循环场景，上述适配代码逻辑是否正确？
B. 代码提示需要Callable<T>，如何修改使其编译通过？

解答 #

问题1：虚拟线程是否适配文档中的线程池用法 #

完全适用。Micrometer的观测上下文传递逻辑对虚拟线程友好，虚拟线程作为JVM层面的轻量级线程，同样支持上下文传递机制，文档中的线程切换观测逻辑可以无缝迁移到虚拟线程场景。

问题2：then(registry.getCurrentObservation()).isSameAs(parent);的用法 #

这是Micrometer Observation的测试断言方法，用于验证线程切换后当前观测上下文是否与父观测一致，仅用于测试场景，生产代码中无需编写该逻辑。

问题3：ContextExecutorService.wrap弃用后的替代方案 #

弃用后推荐使用ObservationAwareExecutorService.wrap(executor)，这是Micrometer专门为观测上下文传递提供的ExecutorService包装类，会自动将当前线程的观测上下文传递到提交的任务中，无论任务运行在普通线程还是虚拟线程。

疑问A：for循环场景下的代码逻辑是否正确 #

原适配代码逻辑存在多处问题：

1. 父Observation未正确启动，createNotStarted创建的观测需通过observe或手动启动，且嵌套方式错误；
2. 每次循环都重新包装ExecutorService，上下文传递逻辑混乱；
3. 子Observation未关联父观测上下文，无法形成完整链路。

正确逻辑应：提前包装线程池，父观测覆盖整个请求流程，子观测自动继承父上下文。

疑问B：解决Callable<T>编译错误 #

编译错误源于lambda返回值不匹配，有两种修改方式：

1. 将子任务改为Runnable类型，去掉返回值：

.submit(() -> {
    Observation.createNotStarted("child", registry)
            .observe(() -> {
                kafkaTemplate.send("topic-loom-micrometer", message + Thread.currentThread().getName());
            });
})

2. 明确指定Callable泛型（如返回Void）：

.submit((Callable<Void>) () -> {
    Observation.createNotStarted("child", registry)
            .observe(() -> {
                kafkaTemplate.send("topic-loom-micrometer", message + Thread.currentThread().getName());
            });
    return null;
})

修正后的完整代码 #

@PostMapping("/question2")
public String question2(@RequestBody List<String> messages) {
    // 包装虚拟线程池，自动传递观测上下文
    try (ExecutorService executor = ObservationAwareExecutorService.wrap(Executors.newVirtualThreadPerTaskExecutor())) {
        // 父观测覆盖整个批量发送流程，自动管理生命周期
        return Observation.createNotStarted("kafka.batch.send", registry)
                .observe(() -> {
                    for (String message : messages) {
                        executor.submit(() -> {
                            // 子观测自动关联父上下文，形成链路
                            Observation.createNotStarted("kafka.message.send", registry)
                                    .observe(() -> {
                                        kafkaTemplate.send("topic-loom-micrometer", message + Thread.currentThread().getName());
                                    });
                        });
                    }
                    executor.shutdown();
                    return "it seems everything happened correctly";
                });
    }
}

代码说明 #

1. ObservationAwareExecutorService.wrap确保任务继承当前观测上下文；
2. 父观测覆盖整个批量发送流程，自动完成启动、结束生命周期管理；
3. 子观测自动关联父上下文，形成完整链路追踪；
4. 子任务使用Runnable类型，解决编译错误。

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