WebFlux + Lettuce Reactive 中 SkyWalking 链路上下文丢失的修复实践

摘要

在 Spring WebFlux 应用中，使用 Lettuce 的 Reactive API 操作 Redis 时，SkyWalking 的分布式追踪常出现链路断裂：Web 请求与 Redis 操作分别产生两条独立的 Trace，TraceId 不一致，导致调用链无法串联。本文记录一次完整的排查与修复过程，从原理到实现，系统梳理 Reactive 场景下链路断裂的真正原因，以及最终的解决方案。目前已提交官方 PR，将在 SkyWalking-Java Agent 9.7 中发布。

一、问题现象

WebFlux + Lettuce Reactive 示例代码如下：

@GetMapping("/lettuce-case")
public Mono<String> lettuceCase() {
    return reactiveStringRedisTemplate.opsForValue().get("key")
            .switchIfEmpty(Mono.just("default"));
}

在 SkyWalking UI 中却看到：

Trace A:
  Spring WebFlux
  ...

Trace B:
  Lettuce/GET

Trace A: Trace B:

二、SkyWalking 跨线程传播机制回顾

要理解问题，需要先回顾 SkyWalking 的上下文模型。

ThreadLocal 模型

在传统的同步或异步回调模型中，SkyWalking 依赖 ThreadLocal 存储链路上下文。ContextManager 内部维护了当前线程的 AbstractTracerContext 和 Span 栈，因此调用链路能自然延续。

跨线程传播

跨线程上下文传播的核心步骤如下：

1. 捕获上下文：调用 ContextManager#capture() 获取 ContextSnapshot 对象；
2. 传递快照：将 ContextSnapshot 对象通过方法参数显式传递到子线程，或存入可跨线程传递的载体（如 Reactor Context）；
3. 恢复上下文：在子线程执行逻辑前，调用 ContextManager#continued(snapshot) 恢复上下文环境。

三、Reactive 模型的根本差异

进入 Reactor 世界后，传播模型完全改变。

Reactor 生命周期

Reactive 执行分为两个阶段：

• assemble（组装阶段） ：构建响应式流，此时所有操作符只是声明式地组合，尚未真正执行。
• subscribe（执行阶段） ：当订阅发生时，数据流才开始实际运行。

关键点：Reactor Context 只有在 subscribe 阶段才可见。

WebFlux Agent 插件已经做了什么

Spring WebFlux 插件已经做了上下文传递的准备工作：

• 在请求入口处捕获 snapshot；
• 将 snapshot 写入 Reactor Context，key 为 SKYWALKING_CONTEXT_SNAPSHOT。

这意味着在 Redis reactive 场景中，上下文其实已经存在于 Reactor Context 中。

四、现有 Lettuce Agent 插件的实现机制

SkyWalking 当前 Lettuce 插件核心拦截点：

4.1 RedisChannelWriter

负责创建出口 Span、设置 peer。该拦截点适用于阻塞模型和异步回调模型。

4.2 RedisCommand 作为增强载体

Lettuce 提供了三种编程模型：同步（RedisCommands）、异步（RedisAsyncCommands）和反应式（RedisReactiveCommands）。无论使用哪种 API，底层最终都会统一抽象为 RedisCommand 对象——它封装了命令的类型、参数以及执行状态。

Skywalking Lettuce Agent 利用这一共性，对 RedisCommand 进行增强，为其附加动态字段，用于在命令生命周期内保存当前追踪的 Span，这一设计在异步回调场景下尤为关键。`

4.3 Reactive 场景为何失效？

createMono()         ← assemble 阶段
  |
RedisPublisher
  |
subscribe()          ← 执行阶段
  |
ChannelWriter

问题根源：关键差异在于执行模型。在阻塞 I/O 模型下，RedisCommand 的创建与执行通常发生在同一调用线程中，TraceContext 基于 ThreadLocal 传播可以自然延续。

而在基于 Project Reactor 的响应式模型中，命令的真正执行发生在 subscribe 阶段，网络 I/O 通常运行在 Netty EventLoop 线程，而非原始业务线程。因此基于 ThreadLocal 的上下文不会自动传播。在这种模型下，上下文需要通过 Reactor Context 显式传播或通过 reactive bridge 进行线程间恢复。

然鹅，SkyWalking Lettuce Agent 插件并未从 Reactor Context 中恢复 snapshot，这导致创建 ExitSpan 时无法获取上游 Span 上下文，最终使得 Redis 调用与 Web 入口的 Trace 链路断裂，无法正确关联。

五、第一次尝试：拦截 createMono（失败）

最直觉的方案是：既然 reactive 从 AbstractRedisReactiveCommands#createMono 开始，就拦截它。目标方法：

public <T> Mono<T> createMono(Supplier<RedisCommand<K, V, T>> commandSupplier)

5.1 失败原因一：RedisCommand 是延迟创建的

createMono 的入参是 Supplier<RedisCommand>，RedisCommand 并未在此时创建，而是在 subscribe → commandSupplier.get() 时才真正实例化。
这就导致了一个根本问题：如果在 afterMethod 中获取 snapshot，此时没有 RedisCommand 实例，无法将 snapshot 绑定到 command 上。

5.2 失败原因二：完全覆写 createMono 不可行

尝试通过 beforeMethod 构造新的 Mono 来覆写原始逻辑，但调试后发现：

• createMono 内部依赖 connection、decorate(commandSupplier)、traceEnabled 等包内访问权限的字段，Agent 无法安全访问；
• 与 Lettuce API 强耦合，版本升级易导致插件失效，违背 SkyWalking 插件设计原则（应基于稳定调用点增强，而非覆写核心逻辑）。

5.3 结论

createMono 不是一个可靠的增强点(这里的Mono并不是最外层的Mono)。必须寻找 执行阶段（subscribe 时刻） 的拦截点。

六、关键突破：找到正确的订阅时机

深入阅读 Lettuce 源码后，关键链路如下：

RedisPublisher.subscribe()
  → new RedisSubscription(connection, command, ...)
  → RedisSubscription.subscribe(subscriber)

此时具备三个关键条件：

• RedisCommand 已创建（通过构造函数传入）；
• CoreSubscriber 可见；
• Reactor Context 可见（通过 subscriber.currentContext()）。

这正是完美的切入点。

七、实现方案设计

但存在一个现实问题：RedisSubscription 内部持有 RedisCommand，却没有公开的 getter 方法。因此需要分两步实现。

八、最终实现：三步拦截注入上下文

Step 1：拦截 RedisSubscription 构造函数

目的：将 RedisCommand 暂存到 SkyWalking 动态字段中，供后续使用。

public class RedisSubscriptionConstructorInterceptor implements InstanceConstructorInterceptor {
    @Override
    public void onConstruct(EnhancedInstance objInst, Object[] allArguments) {
        // allArguments[1] 正是 RedisCommand 实例
        objInst.setSkyWalkingDynamicField(allArguments[1]);
    }
}

Step 2：拦截 subscribe 方法

在 subscribe 阶段，从 Reactor Context 中获取 snapshot，并绑定到 RedisCommand 的增强实例上。

public class RedisSubscriptionSubscribeMethodInterceptor implements InstanceMethodsAroundInterceptorV2 {
    @Override
    public void beforeMethod(EnhancedInstance objInst,
                             Method method,
                             Object[] allArguments,
                             Class<?>[] argumentsTypes,
                             MethodInvocationContext context) {
        if (allArguments[0] instanceof CoreSubscriber) {
            CoreSubscriber<?> subscriber = (CoreSubscriber<?>) allArguments[0];
            // 从 Reactor Context 中获取 snapshot
            Object snapshot = subscriber.currentContext()
                              .getOrDefault("SKYWALKING_CONTEXT_SNAPSHOT", null);
            if (snapshot != null) {
                // 将 snapshot 设置到 RedisCommand 的动态字段中
                ((EnhancedInstance) objInst.getSkyWalkingDynamicField())
                    .setSkyWalkingDynamicField(
                        new RedisCommandEnhanceInfo().setSnapshot((ContextSnapshot) snapshot));
            }
        }
    }
}

至此，RedisCommand 的动态字段中已经包含了从 Reactor Context 传递过来的上下文快照。

Step 3：增强 RedisChannelWriter 拦截器

原有的 RedisChannelWriter 拦截点负责创建出口 Span，但此前它并不知道上下文的存在。现在需要修改该拦截器，在执行真正写入操作前检查 RedisCommand 中是否携带了 snapshot，如果有则先恢复上下文，再创建 Span。

关键修改位于 RedisChannelWriter 拦截器的 beforeMethod 中：

// 从 RedisCommand 增强实例中获取之前存入的增强信息
RedisCommandEnhanceInfo enhanceInfo = (RedisCommandEnhanceInfo) ((EnhancedInstance) redisCommand).getSkyWalkingDynamicField();
if (enhanceInfo != null && enhanceInfo.getSnapshot() != null) {
    // 创建本地 Span 并恢复上下文，保证后续 ExitSpan 能正确关联
    AbstractSpan localSpan = ContextManager.createLocalSpan("RedisReactive/local");
    localSpan.setComponent(ComponentsDefine.LETTUCE);
    SpanLayer.asCache(localSpan);
    ContextManager.continued(enhanceInfo.getSnapshot());
}
// 然后继续原有逻辑：创建 ExitSpan、设置 peer、记录命令等

这段代码的作用：

• 从 RedisCommand 的动态字段中取出 snapshot；
• 创建一个本地 Span（用于标识上下文恢复点）；
• 调用 ContextManager.continued(snapshot) 将快照中的上下文绑定到当前线程；
• 之后原有逻辑创建的 ExitSpan 就会自动归属到正确的 Trace 中。

九、最终链路结构

WebFlux Entry
   ↓
Reactor Context (snapshot)
   ↓
RedisSubscription.subscribe  ⭐ (Step2: snapshot → RedisCommand)
   ↓
RedisCommand (enhanced with snapshot)
   ↓
RedisChannelWriter.beforeMethod  ⭐ (Step3: continued + ExitSpan)
   ↓
ExitSpan

结果：

• Web 与 Redis 完整串联；
• TraceId 一致；

十、总结

这次问题的本质是 ThreadLocal 思维与 Reactive 思维的错位。在传统的同步/异步模型中，上下文依赖线程绑定；而在 Reactive 世界中，执行被拆分为组装与订阅两个阶段，上下文必须通过 Reactor Context 显式传递。理解这一根本差异，才能找到正确的修复切入点。

关键经验：

1. 理解执行阶段是前提：Reactive 插件的增强点必须选在订阅（subscribe）时刻，此时才能访问 Reactor Context 中的上下文快照。
2. 上下文传递载体：必须依托 Reactor Context 传递 snapshot；
3. 警惕 Supplier 延迟创建：当 API 使用 Supplier 延迟创建对象时，不能简单在创建处绑定上下文；
4. 遵循插件设计原则：不要覆写核心 API，应利用现有增强机制（如动态字段）；

修复完成后，我按照 SkyWalking 官方的插件测试规范补充了对应的E2E测试用例，覆盖 Spring WebFlux 5.x/6.x + Lettuce Reactive 的完整调用链场景。

目前该修复方案已通过 Pull Request 提交给 SkyWalking 社区，经过 Member 的 review 与讨论后成功合入主干。预计将在 SkyWalking-Java Agent 9.7 版本中正式发布。相关 PR 及讨论记录可参考：apache/skywalking-java#788