带有 CompletableFuture 的 MDC 记录器

Question

我正在使用 MDC 记录器，除了一种情况外，它对我来说非常有用。无论我们在代码中的什么地方使用了 CompletableFuture，对于创建的线程，MDC 数据都不会传递到下一个线程，因此日志失败。例如，在我使用以下代码片段创建新线程的代码中。

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 数据，它没有通过 Thread [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>

我在下面试过 Link

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+ 一起使用，因为自该版本以来公开了几个可覆盖的方法）

Make the complete ecosystem aware of MDC

为此，我们需要解决以下情况：

我们什么时候才能从这个 class 中获得 CompletableFuture 的新实例？ → 我们需要 return 一个 MDC 感知版本.
我们什么时候才能从外部获得 CompletableFuture 的新实例 class? → 我们需要 return 相同的 MDC 感知版本.
在 CompletableFuture 中使用哪个执行器 class? → 在所有情况下，我们需要确保所有执行器都是 MDC 感知的

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

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 class 看起来像（为简单起见跳过了带有 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);
    }
}

以下是一些使用指南：

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

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

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

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

带有 CompletableFuture 的 MDC 记录器

MDC Logger with CompletableFuture

java

logback

slf4j

mdc

completable-future