具有 CompletableFuture 的 MDC 记录器

Question

我正在使用 MDC Logger，它对我来说非常有用，但有一种情况除外。在我们使用 CompletableFuture 的代码中，对于创建的线程，MDC 数据不会传递到下一个线程，因此日志失败。例如，在我在 sn-p 下面用于创建新线程的代码中。

CompletableFuture.runAsync(() -> getAcountDetails(user));

日志结果如下

2019-04-29 11:44:13,690 INFO  | /app/rest/controller/userdetails | f80fdc1f-8123-3932-a405-dda2dc2a80d5 |[http-nio-8182-exec-5] RestServiceExecutor:  service: 
2019-04-29 11:44:13,690 INFO  | /app/rest/controller/userdetails | f80fdc1f-8123-3932-a405-dda2dc2a80d5 |[http-nio-8182-exec-5] RestServiceExecutor: 
2019-04-29 11:44:13,779 INFO  | /app/rest/controller/userdetails | f80fdc1f-8123-3932-a405-dda2dc2a80d5 |[http-nio-8182-exec-5] UserDetailsRepoImpl: 
2019-04-29 11:44:13,950 INFO   [ForkJoinPool.commonPool-worker-3] RestServiceExecutor:  header: 
2019-04-29 11:44:13,950 INFO   [ForkJoinPool.commonPool-worker-3] RestServiceExecutor:  service: 
2019-04-29 11:44:14,012 INFO   [ForkJoinPool.commonPool-worker-3] CommonMasterDataServiceImpl: Cache: Retrieving Config Data details.
2019-04-29 11:44:14,028 INFO   [ForkJoinPool.commonPool-worker-3] CommonMasterDataServiceImpl: Cache: Retrieved Config Data details : 1
2019-04-29 11:44:14,028 INFO   [ForkJoinPool.commonPool-worker-3] CommonMasterDataServiceImpl: Cache: Retrieving Config Data details.
2019-04-29 11:44:14,033 INFO   [ForkJoinPool.commonPool-worker-3] CommonMasterDataServiceImpl: Cache: Retrieved Config Data details : 1
2019-04-29 11:44:14,147 INFO  | /app/rest/controller/userdetails | f80fdc1f-8123-3932-a405-dda2dc2a80d5 |[http-nio-8182-exec-5] SecondaryCacheServiceImpl: Fetching from secondary cache
2019-04-29 11:44:14,715 INFO  | /app/rest/controller/userdetails | f80fdc1f-8123-3932-a405-dda2dc2a80d5 |[http-nio-8182-exec-5] CommonMasterDataServiceImpl: Cache: Retrieving Config Data details.
2019-04-29 11:44:14,749 INFO  | /app/rest/controller/userdetails | f80fdc1f-8123-3932-a405-dda2dc2a80d5 |[http-nio-8182-exec-5]

下面是我的 MDC 数据，它没有通过线程 [ForkJoinPool.commonPool-worker-3] 传递

| /app/rest/controller/userdetails | f80fdc1f-8123-3932-a405-dda2dc2a80d5 |

下面是我的 logback.xml 配置，其中 sessionID 是 MDC 数据

<configuration scan="true">
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <charset>utf-8</charset>
            <Pattern>%d %-5level %X{sessionID} [%thread] %logger{0}: %msg%n</Pattern>
        </encoder>
    </appender>
</configuration>

我试过下面的链接

http://shengwangi.blogspot.com/2015/09/using-log-mdc-in-multi-thread-helloworld-example.html?_sm_au_=iVVrZDSwwf0vP6MR

这非常适合 TaskExecutor。但是我还没有找到 CompletableFuture 的任何解决方案。

Answer 1

创建包装方法

static CompletableFuture<Void> myMethod(Runnable runnable) {
    Map<String, String> previous = MDC.getCopyOfContextMap();
    return CompletableFuture.runAsync(() -> {
        MDC.setContextMap(previous);
        try {
            runnable.run();
        } finally {
            MDC.clear();
        }
    });
}

并使用它来代替CompletableFuture.runAsync。

Answer 2

我的解决方案主题是（它可以与 JDK 9+ 一起使用，因为自该版本以来公开了几个可覆盖的方法）

让整个生态系统了解 MDC

为此，我们需要解决以下场景：

• 什么时候我们可以从这个类中获得 CompletableFuture 的新实例？ → 我们需要返回一个 MDC 感知版本。
• 什么时候我们才能从这个类之外获得 CompletableFuture 的新实例？ → 我们需要返回相同的 MDC 感知版本。
• 在 CompletableFuture 类中使用哪个执行器？ → 在所有情况下，我们都需要确保所有执行器都支持 MDC

为此，让我们通过扩展它来创建CompletableFuture 的MDC 感知版本类。我的版本如下所示

import org.slf4j.MDC;

import java.util.Map;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.function.Supplier;

public class MDCAwareCompletableFuture<T> extends CompletableFuture<T> {

    public static final ExecutorService MDC_AWARE_ASYNC_POOL = new MDCAwareForkJoinPool();

    @Override
    public CompletableFuture newIncompleteFuture() {
        return new MDCAwareCompletableFuture();
    }

    @Override
    public Executor defaultExecutor() {
        return MDC_AWARE_ASYNC_POOL;
    }

    public static <T> CompletionStage<T> getMDCAwareCompletionStage(CompletableFuture<T> future) {
        return new MDCAwareCompletableFuture<>()
                .completeAsync(() -> null)
                .thenCombineAsync(future, (aVoid, value) -> value);
    }

    public static <T> CompletionStage<T> getMDCHandledCompletionStage(CompletableFuture<T> future,
                                                                Function<Throwable, T> throwableFunction) {
        Map<String, String> contextMap = MDC.getCopyOfContextMap();
        return getMDCAwareCompletionStage(future)
                .handle((value, throwable) -> {
                    setMDCContext(contextMap);
                    if (throwable != null) {
                        return throwableFunction.apply(throwable);
                    }
                    return value;
                });
    }
}

MDCAwareForkJoinPool 类看起来像（为简单起见，跳过了带有 ForkJoinTask 参数的方法）

public class MDCAwareForkJoinPool extends ForkJoinPool {
    //Override constructors which you need

    @Override
    public <T> ForkJoinTask<T> submit(Callable<T> task) {
        return super.submit(MDCUtility.wrapWithMdcContext(task));
    }

    @Override
    public <T> ForkJoinTask<T> submit(Runnable task, T result) {
        return super.submit(wrapWithMdcContext(task), result);
    }

    @Override
    public ForkJoinTask<?> submit(Runnable task) {
        return super.submit(wrapWithMdcContext(task));
    }

    @Override
    public void execute(Runnable task) {
        super.execute(wrapWithMdcContext(task));
    }
}

包装的实用方法是这样的

public static <T> Callable<T> wrapWithMdcContext(Callable<T> task) {
    //save the current MDC context
    Map<String, String> contextMap = MDC.getCopyOfContextMap();
    return () -> {
        setMDCContext(contextMap);
        try {
            return task.call();
        } finally {
            // once the task is complete, clear MDC
            MDC.clear();
        }
    };
}

public static Runnable wrapWithMdcContext(Runnable task) {
    //save the current MDC context
    Map<String, String> contextMap = MDC.getCopyOfContextMap();
    return () -> {
        setMDCContext(contextMap);
        try {
            task.run();
        } finally {
            // once the task is complete, clear MDC
            MDC.clear();
        }
    };
}

public static void setMDCContext(Map<String, String> contextMap) {
   MDC.clear();
   if (contextMap != null) {
       MDC.setContextMap(contextMap);
    }
}

以下是一些使用指南：

• 使用MDCAwareCompletableFuture 类而不是CompletableFuture 类。
• CompletableFuture 类中的几个方法实例化了 self 版本，例如 new CompletableFuture...。对于此类方法（大多数公共静态方法），使用替代方法获取MDCAwareCompletableFuture 的实例。使用替代方法的示例可能不是使用CompletableFuture.supplyAsync(...)，您可以选择new MDCAwareCompletableFuture<>().completeAsync(...)
• 当您因为某个外部库返回CompletableFuture 的实例而遇到困难时，请使用getMDCAwareCompletionStage 方法将CompletableFuture 的实例转换为MDCAwareCompletableFuture。显然，您不能在该库中保留上下文，但在您的代码命中应用程序代码后，此方法仍会保留上下文。
• 在提供执行程序作为参数时，请确保它是 MDC 感知的，例如 MDCAwareForkJoinPool。您也可以通过覆盖 execute 方法来创建 MDCAwareThreadPoolExecutor 以服务于您的用例。你明白了！

您可以在post 中找到上述所有内容的详细解释。

这样，您的代码可能看起来像

new MDCAwareCompletableFuture<>().completeAsync(() -> {
            getAcountDetails(user);
            return null;
        });