设想场景

假如我们需要跟踪某条请求的所有后台日志，其中这些日志的埋点有同步的，也有异步的，甚至是使用Reactor的，那这个时候，我们应该怎么跟踪？这个在分布式服务和微服务下叫全链路监控-APM，现在我们就在单机环境下即同一jvm下说明这个问题。

同步线程

SLF4J 日志框架提供了一个 MDC(Mapped Diagnostic Contexts) 工具类

public class Main {

    private static final String KEY = "requestId";
    private static final Logger logger = LoggerFactory.getLogger(Main.class);
    
    public static void main(String[] args) {

        // 入口传入请求ID
        MDC.put(KEY, UUID.randomUUID().toString());
        
        // 打印日志
        logger.debug("log in main thread 1");
        logger.debug("log in main thread 2");
        logger.debug("log in main thread 3");

        // 出口移除请求ID
        MDC.remove(KEY);
    }
}

具体可以参考如何快速过滤出一次请求的所有日志

异步线程

由于 Logback 的 MDC 实际上是一个 ThreadLocal 的实现，因此，当异步执行产生线程切换时，需要将 MDC 保存的信息进行切换。
Spring 中有一个可用的线程装饰器TaskDecorator，这个是 Spring Core 4.3 版本才加入的接口，通过实现这个接口，可以自己控制传播那些变量

/**
 * 解决异步执行时MDC内容延续的问题
 */
public class MDCTaskDecorator implements TaskDecorator {
    
    @Override
    public Runnable decorate(Runnable runnable) {
        return new MDCContinueRunableDecorator(runnable);
    }
    
    /**
     * 执行线程装饰器
     */
    protected class MDCContinueRunableDecorator implements Runnable {
        
        private final Runnable delegate;
        
        protected final Map<String, String> logContextMap;
        
        public MDCContinueRunableDecorator(Runnable runnable) {
            this.delegate = runnable;
            this.logContextMap = MDC.getCopyOfContextMap();
        }
        
        @Override
        public void run() {
            MDC.setContextMap(this.logContextMap);
            this.delegate.run();
            MDC.clear();
        }
    }
}

然后，需要自定义实现一个 TaskExecutor，替换 Spring 提供的默认实现，代码如下。

 /**
     * 自定义线程池
     * <p>
     * 用于线程切换时的MDC延续
     */
    @Bean
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(maxPoolSize);
        executor.setMaxPoolSize(maxPoolSize);
        executor.setTaskDecorator(new MDCTaskDecorator());
        executor.setThreadNamePrefix("MDCAdaptTaskExcutor-");
        executor.initialize();
        return executor;
    }

只要异步处理使用了自定义的 TaskExecutor ，即可实现上下文的自动传递。

Reactor

spring5引入webflux，其底层是基于reactor，那么reactor如何进行上下文变量的传播呢？官方提供了Context对象来替代threadlocal。

其特性如下：

类似map的kv操作，比如put(Object key, Object value),putAll(Context), hasKey(Object key)
immutable，即同一个key，后面put不会覆盖
提供getOrDefault，getOrEmpty方法
Context与作用链上的每个Subscriber绑定
通过subscriberContext(Context)来访问
Context的作用是自底向上

实例

设置及读取

    @Test
    public void testSubscriberContext(){
        String key = "message";
        Mono<String> r = Mono.just("Hello")
                .flatMap( s -> Mono.subscriberContext()
                        .map( ctx -> s + " " + ctx.get(key)))
                .subscriberContext(ctx -> ctx.put(key, "World"));

        StepVerifier.create(r)
                .expectNext("Hello World")
                .verifyComplete();
    }

这里从最底部的subscriberContext设置message值为World，然后flatMap里头通过subscriberContext来访问。

自底向上

    @Test
    public void testContextSequence(){
        String key = "message";
        Mono<String> r = Mono.just("Hello")
                //NOTE 这个subscriberContext设置的太高了
                .subscriberContext(ctx -> ctx.put(key, "World"))
                .flatMap( s -> Mono.subscriberContext()
                        .map( ctx -> s + " " + ctx.getOrDefault(key, "Stranger")));

        StepVerifier.create(r)
                .expectNext("Hello Stranger")
                .verifyComplete();
    }
复制代码

由于这个例子的subscriberContext设置的太高了，不能作用在flatMap里头的Mono.subscriberContext()

不可变

    @Test
    public void testContextImmutable(){
        String key = "message";

        Mono<String> r = Mono.subscriberContext()
                .map( ctx -> ctx.put(key, "Hello"))
                //这里返回了一个新的，因此上面的设置失效了
                .flatMap( ctx -> Mono.subscriberContext())
                .map( ctx -> ctx.getOrDefault(key,"Default"));

        StepVerifier.create(r)
                .expectNext("Default")
                .verifyComplete();
    }

subscriberContext永远返回一个新的

多个连续的subscriberContext

    @Test
    public void testReadOrder(){
        String key = "message";
        Mono<String> r = Mono.just("Hello")
                .flatMap( s -> Mono.subscriberContext()
                        .map( ctx -> s + " " + ctx.get(key)))
                .subscriberContext(ctx -> ctx.put(key, "Reactor"))
                .subscriberContext(ctx -> ctx.put(key, "World"));

        StepVerifier.create(r)
                .expectNext("Hello Reactor")
                .verifyComplete();
    }

operator只会读取离它最近的一个context

flatMap间的subscriberContext

    @Test
    public void testContextBetweenFlatMap(){
        String key = "message";
        Mono<String> r = Mono.just("Hello")
                .flatMap( s -> Mono.subscriberContext()
                        .map( ctx -> s + " " + ctx.get(key)))
                .subscriberContext(ctx -> ctx.put(key, "Reactor"))
                .flatMap( s -> Mono.subscriberContext()
                        .map( ctx -> s + " " + ctx.get(key)))
                .subscriberContext(ctx -> ctx.put(key, "World"));

        StepVerifier.create(r)
                .expectNext("Hello Reactor World")
                .verifyComplete();
    }

flatMap读取离它最近的context

flatMap中的subscriberContext

    @Test
    public void testContextInFlatMap(){
        String key = "message";
        Mono<String> r =
                Mono.just("Hello")
                        .flatMap( s -> Mono.subscriberContext()
                                .map( ctx -> s + " " + ctx.get(key))
                        )
                        .flatMap( s -> Mono.subscriberContext()
                                .map( ctx -> s + " " + ctx.get(key))
                                .subscriberContext(ctx -> ctx.put(key, "Reactor"))
                        )
                        .subscriberContext(ctx -> ctx.put(key, "World"));

        StepVerifier.create(r)
                .expectNext("Hello World Reactor")
                .verifyComplete();
    }