Spring Cloud Gateway 没有链路信息，我 TM 人傻了（下） _gateway

努力尽今夕，少年犹可夸。这篇文章主要讲述Spring Cloud Gateway 没有链路信息，我 TM 人傻了（下）相关的知识，希望能为你提供帮助。

文章图片

本篇文章涉及底层设计以及原理，以及问题定位和可能的问题点，非常深入，篇幅较长，所以拆分成上中下三篇：

上：问题简单描述以及 Spring Cloud Gateway 基本结构和流程以及底层原理
中：Spring Cloud Sleuth 如何在 Spring Cloud Gateway 加入的链路追踪以及为何会出现这个问题
下：现有 Spring Cloud Sleuth 的非侵入设计带来的性能问题，其他可能的问题点，以及如何解决

Spring Cloud Gateway 其他的可能丢失链路信息的点【Spring Cloud Gateway 没有链路信息，我 TM 人傻了（下）】经过前面的分析，我们可以看出，不止这里，还有其他地方会导致 Spring Cloud Sleuth 的链路追踪信息消失，这里举几个大家常见的例子：
1.在 GatewayFilter 中指定了异步执行某些任务，由于线程切换了，并且这时候可能 Span 已经结束了，所以没有链路信息，例如：

@Override public Mono< Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) { return chain.filter(exchange).publishOn(Schedulers.parallel()).doOnSuccess(o -> { //这里就没有链路信息了 log.info("success"); }); }

2.将 GatewayFilter 中继续链路的 chain.filter(exchange) 放到了异步任务中执行，上面的 AdaptCachedBodyGlobalFilter 就属于这种情况，这样会导致之后的 GatewayFilter 都没有链路信息，例如：

@Override public Mono< Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) { return Mono.delay(Duration.ofSeconds(1)).then(chain.filter(exchange)); }

java 并发编程模型与 Project Reactor 编程模型的冲突思考Java 中的很多框架，都用到了 ThreadLocal，或者通过 Thread 来标识唯一性。例如：

日志框架中的 MDC，一般都是 ThreadLocal 实现。
所有的锁、基于 AQS 的数据结构，都是通过 Thread 的属性来唯一标识谁获取到了锁的。
分布式锁等数据结构，也是通过 Thread 的属性来唯一标识谁获取到了锁的，例如 Redisson 中分布式 Redis 锁的实现。

但是放到 Project Reactor 编程模型，这就显得格格不入了，因为 Project Reactor 异步响应式编程就是不固定线程，没法保证提交任务和回调能在同一个线程，所以 ThreadLocal 的语义在这里很难成立。Project Reactor 虽然提供了对标 ThreadLocal 的 Context，但是主流框架还没有兼容这个 Context，所以给 Spring Cloud Sleuth 粘合这些链路追踪带来了很大困难，因为 MDC 是一个 ThreadLocal 的 Map 实现，而不是基于 Context 的 Map。这就需要 Spring Cloud Sleuth 在订阅一开始，就需要将链路信息放入 MDC，同时还需要保证运行时不切换线程。
运行不切换线程，这样其实限制了 Project Reactor 的灵活调度，是有一些性能损失的。我们其实想尽量就算加入了链路追踪信息，也不用强制运行不切换线程。但是 Spring Cloud Sleuth 是非侵入式设计，很难实现这一点。但是对于我们自己业务的使用，我们可以定制一些编程规范，来保证大家写的代码不丢失链路信息。
改进我们的编程规范首先，我们自定义 Mono 和 Flux 的工厂
公共 Subscriber 封装，将 reactor Subscriber 的所有关键接口，都检查当前上下文是否有链路信息，即 Span，如果没有就包裹上，如果有则直接执行即可。

public class TracedCoreSubscriber< T> implements Subscriber< T> { private final Subscriber< T> delegate; private final Tracer tracer; private final CurrentTraceContext currentTraceContext; private final Span span; TracedCoreSubscriber(Subscriber< T> delegate, Tracer tracer, CurrentTraceContext currentTraceContext, Span span) { this.delegate = delegate; this.tracer = tracer; this.currentTraceContext = currentTraceContext; this.span = span; }@Override public void onSubscribe(Subscription s) { executeWithinScope(() -> { delegate.onSubscribe(s); }); }@Override public void onError(Throwable t) { executeWithinScope(() -> { delegate.onError(t); }); }@Override public void onComplete() { executeWithinScope(() -> { delegate.onComplete(); }); }@Override public void onNext(T o) { executeWithinScope(() -> { delegate.onNext(o); }); }private void executeWithinScope(Runnable runnable) { //如果当前没有链路信息，强制包裹 if (tracer.currentSpan() == null) { try (CurrentTraceContext.Scope scope = this.currentTraceContext.maybeScope(this.span.context())) { runnable.run(); } } else { //如果当前已有链路信息，则直接执行 runnable.run(); } } }

之后分别定义所有 Flux 的代理 TracedFlux，和所有 Mono 的代理 TracedMono，其实就是在 subscribe 的时候，用 TracedCoreSubscriber 包装传入的 CoreSubscriber：

public class TracedFlux< T> extends Flux< T> { private final Flux< T> delegate; private final Tracer tracer; private final CurrentTraceContext currentTraceContext; private final Span span; TracedFlux(Flux< T> delegate, Tracer tracer, CurrentTraceContext currentTraceContext, Span span) { this.delegate = delegate; this.tracer = tracer; this.currentTraceContext = currentTraceContext; this.span = span; }@Override public void subscribe(CoreSubscriber< ? super T> actual) { delegate.subscribe(new TracedCoreSubscriber(actual, tracer, currentTraceContext, span)); } }public class TracedMono< T> extends Mono< T> { private final Mono< T> delegate; private final Tracer tracer; private final CurrentTraceContext currentTraceContext; private final Span span; TracedMono(Mono< T> delegate, Tracer tracer, CurrentTraceContext currentTraceContext, Span span) { this.delegate = delegate; this.tracer = tracer; this.currentTraceContext = currentTraceContext; this.span = span; }@Override public void subscribe(CoreSubscriber< ? super T> actual) { delegate.subscribe(new TracedCoreSubscriber(actual, tracer, currentTraceContext, span)); } }

定义工厂类，使用请求 ServerWebExchange 和原始 Flux 创建 TracedFlux，以及使用请求 ServerWebExchange 和原始 Mono 创建 TracedMono，并且 Span 是通过 Attributes 获取的，根据前文的源码分析我们知道，这个 Attribute 是通过 TraceWebFilter 放入 Attributes 的。由于我们只在 GatewayFilter 中使用，一定在 TraceWebFilter 之后所以这个 Attribute 一定存在。

@Component public class TracedPublisherFactory { protected static final String TRACE_REQUEST_ATTR = Span.class.getName(); @Autowired private Tracer tracer; @Autowired private CurrentTraceContext currentTraceContext; public < T> Flux< T> getTracedFlux(Flux< T> publisher, ServerWebExchange exchange) { return new TracedFlux< > (publisher, tracer, currentTraceContext, (Span) exchange.getAttributes().get(TRACE_REQUEST_ATTR)); }public < T> Mono< T> getTracedMono(Mono< T> publisher, ServerWebExchange exchange) { return new TracedMono< > (publisher, tracer, currentTraceContext, (Span) exchange.getAttributes().get(TRACE_REQUEST_ATTR)); } }

然后，我们规定：1. 所有的 GatewayFilter，需要继承我们自定义的抽象类，这个抽象类仅仅是把 filter 的结果用 TracedPublisherFactory 的 getTracedMono 给封装了一层 TracedMono，以 GlobalFilter 为例子：

public abstract class AbstractTracedFilter implements GlobalFilter { @Autowired protected TracedPublisherFactory tracedPublisherFactory; @Override public Mono< Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) { return tracedPublisherFactory.getTracedMono(traced(exchange, chain), exchange); }protected abstract Mono< Void> traced(ServerWebExchange exchange, GatewayFilterChain chain); }

2. GatewayFilter 中新生成的 Flux 或者 Mono，统一使用 TracedPublisherFactory 再封装一层。
3. 对于 AdaptCachedBodyGlobalFilter 读取 Request Body 导致的链路丢失，我向社区提了一个 Pull Request: fix #2004 Span is not terminated properly in Spring Cloud Gateway，大家可以参考。也可以在这个 Filter 之前自己将 Request Body 使用 TracedPublisherFactory 进行封装解决。