SpringCloud升级之路2020.0.x版-43.为何 SpringCloudGateway 中会有链路信息丢失

<parent>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-parent</artifactId>
	<version>2.4.6</version>
</parent>

<dependencies>
	<dependency>
		<groupId>org.springframework.cloud</groupId>
		<artifactId>spring-cloud-starter-gateway</artifactId>
	</dependency>
	<dependency>
		<groupId>org.springframework.cloud</groupId>
		<artifactId>spring-cloud-starter-sleuth</artifactId>
	</dependency>
	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-log4j2</artifactId>
	</dependency>
	<!--log4j2异步日志需要的依赖，所有项目都必须用log4j2和异步日志配置-->
	<dependency>
		<groupId>com.lmax</groupId>
		<artifactId>disruptor</artifactId>
		<version>${disruptor.version}</version>
	</dependency>
</dependencies>

<dependencyManagement>
	<dependencies>
		<dependency>
			<groupId>org.springframework.cloud</groupId>
			<artifactId>spring-cloud-dependencies</artifactId>
			<version>2020.0.3</version>
			<type>pom</type>
			<scope>import</scope>
		</dependency>
	</dependencies>
</dependencyManagement>

@Configuration(proxyBeanMethods = false)
public class ApiGatewayConfiguration {
    @Autowired
    private AdaptCachedBodyGlobalFilter adaptCachedBodyGlobalFilter;
    @Autowired
    private GatewayProperties gatewayProperties;

    @PostConstruct
    public void init() {
        gatewayProperties.getRoutes().forEach(routeDefinition -> {
			//对 spring cloud gateway 路由配置中的每个路由都启用 AdaptCachedBodyGlobalFilter 
            EnableBodyCachingEvent enableBodyCachingEvent = new EnableBodyCachingEvent(new Object(), routeDefinition.getId());
            adaptCachedBodyGlobalFilter.onApplicationEvent(enableBodyCachingEvent);
        });
    }
}

server:
  port: 8181
spring:
  application:
    name: apiGateway
  cloud:
    gateway:
      httpclient:
        connect-timeout: 500
        response-timeout: 60000
      routes:
        - id: first_route
          uri: http://httpbin.org
          predicates:
              - Path=/httpbin/**
          filters:
              - StripPrefix=1

@Log4j2
@Component
public class PreLogFilter implements GlobalFilter, Ordered {

    public static final int ORDER = new AdaptCachedBodyGlobalFilter().getOrder() - 1;

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        log.info("before AdaptCachedBodyGlobalFilter");
        return chain.filter(exchange);
    }

    @Override
    public int getOrder() {
        return ORDER;
    }
}

@Log4j2
@Component
public class PostLogFilter implements GlobalFilter, Ordered {

    public static final int ORDER = new AdaptCachedBodyGlobalFilter().getOrder() + 1;

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        log.info("after AdaptCachedBodyGlobalFilter");
        return chain.filter(exchange);
    }

    @Override
    public int getOrder() {
        return ORDER;
    }
}

2021-09-08 06:32:35.457  INFO [service-apiGateway,51063d6f1fe264d0,51063d6f1fe264d0] [30600] [reactor-http-nio-2][?:]: before AdaptCachedBodyGlobalFilter
2021-09-08 06:32:35.474  INFO [service-apiGateway,,] [30600] [reactor-http-nio-2][?:]: after AdaptCachedBodyGlobalFilter

@Override
public void subscribe(CoreSubscriber<? super Void> subscriber) {
	Context context = contextWithoutInitialSpan(subscriber.currentContext());
	Span span = findOrCreateSpan(context);
	//将 Span 放入执行上下文中，对于日志其实就是将链路信息放入 org.slf4j.MDC
	//日志的 MDC 一般都是 ThreadLocal 的 Map，对于 Log4j2 的实现类就是 org.apache.logging.log4j.ThreadContext，其核心 contextMap 就是一个基于 ThreadLocal 实现的 Map
	//简单理解就是将链路信息放入一个 ThreadLocal 的 Map 中，每个线程访问自己的 Map 获取链路信息
	try (CurrentTraceContext.Scope scope = this.currentTraceContext.maybeScope(span.context())) {
		//将实际的 subscribe 用 Span 所在的 Context 包裹住，结束时关闭 Span
		this.source.subscribe(new WebFilterTraceSubscriber(subscriber, context, span, this));
	}
	//在 scope.close() 之后，会将链路信息从  ThreadLocal 的 Map 中剔除
}

@Override
public Object scanUnsafe(Attr key) {
	if (key == Attr.RUN_STYLE) {
		//执行的方式必须是不能切换线程，也就是同步的
		//因为，日志的链路信息是放在 ThreadLocal 对象中，切换线程，链路信息就没了
		return Attr.RunStyle.SYNC; 
	}
	return super.scanUnsafe(key);
}

return Mono.defer(() ->
	new MonoWebFilterTrace(source, 
		RoutePredicateHandlerMapping.this.lookupRoute(exchange) //根据请求寻找路由
				.flatMap((Function<Route, Mono<?>>) r -> {
					exchange.getAttributes().put(GATEWAY_ROUTE_ATTR, r); //将路由放入 Attributes 中，后面我们还会用到
					return Mono.just(RoutePredicateHandlerMapping.this.webHandler); //返回 RoutePredicateHandlerMapping 的 FilteringWebHandler
				}).switchIfEmpty( //如果为 Mono.empty()，也就是没找到路由
					Mono.empty() 
					.then(Mono.fromRunnable(() -> { //返回 Mono.empty() 之后，记录日志
						if (logger.isTraceEnabled()) {
							logger.trace("No RouteDefinition found for [" + getExchangeDesc(exchange) + "]");
					}
				})))
			.switchIfEmpty(DispatcherHandler.this.createNotFoundError()) //如果没有返回不为 Mono.empty() 的 handlerMapping，则直接返回 404
			.then(
				Mono.defer(() -> {
					//省略在 AdaptCachedBodyGlobalFilter 前面的链路嵌套
					//读取 Body，由于 TCP 拆包，所以需要他们拼接到一起
					DataBufferUtils.join(exchange.getRequest().getBody())
						//如果没有 Body，则直接返回空 DataBuffer
						.defaultIfEmpty(factory.wrap(new EmptyByteBuf(factory.getByteBufAllocator())))
						//decorate方法中将 dataBuffer 放入 exchange 的 Attributes 列表，只是为了防止重复进入这个 `AdaptCachedBodyGlobalFilter` 的情况导致重复缓存请求 Body
						//之后，使用新的 body 以及原始请求封装成新的请求，继续 GatewayFilters 链路
						.map(dataBuffer -> decorate(exchange, dataBuffer, cacheDecoratedRequest))
						.switchIfEmpty(Mono.just(exchange.getRequest())).flatMap(function);
				})
				.then(Mono.empty()))
			), //调用对应的 Handler
	TraceWebFilter.this.isTracePresent(), TraceWebFilter.this, TraceWebFilter.this.spanFromContextRetriever()).transformDeferred((call) -> {
		//MetricsWebFilter 相关的处理，在前面的代码中给出了，这里省略
	});
);

//首先我们创建一个新的 Span
Span span = tracer.newTrace();
//声明一个类似于 TraceWebFilter 中封装的 MonoWebFilterTrace 的 MonoOperator
class MonoWebFilterTrace<T> extends MonoOperator<T, T> {
	protected MonoWebFilterTrace(Mono<? extends T> source) {
		super(source);
	}

	@Override
	public void subscribe(CoreSubscriber<? super T> actual) {
		//将 subscribe 用 span 包裹
		try (Tracer.SpanInScope spanInScope = tracer.withSpanInScope(span)) {
			source.subscribe(actual);
			//在将要关闭 spanInScope 的时候（即从 ThreadLocal 的 Map 中移除链路信息），打印日志
			log.info("stopped");
		}
	}
}

Mono.defer(() -> new MonoWebFilterTrace(
		Mono.fromRunnable(() -> {
			log.info("first");
		})
		//模拟 FluxReceive
		.then(Mono.delay(Duration.ofSeconds(1))
		.doOnSuccess(longSignal -> log.info(longSignal))))
).subscribe(aLong -> log.info(aLong));

2021-09-08 07:12:45.236  INFO [service-apiGateway,7b2f5c190e1406cb,7b2f5c190e1406cb] [31868] [reactor-http-nio-2][?:]: first
2021-09-08 07:12:45.240  INFO [service-apiGateway,7b2f5c190e1406cb,7b2f5c190e1406cb] [31868] [reactor-http-nio-2][?:]: stopped
2021-09-08 07:12:46.241  INFO [service-apiGateway,,] [31868] [parallel-1][?:]: doOnEach_onNext(0)
2021-09-08 07:12:46.242  INFO [service-apiGateway,,] [31868] [parallel-1][?:]: onComplete()
2021-09-08 07:12:46.242  INFO [service-apiGateway,,] [31868] [parallel-1][?:]: 0