具有 CompletableFuture 的 MDC 记录器

Question

我正在使用 MDC Logger，除了一种情况外，它对我来说非常有用。在我们使用 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 : 12019-04-29 11:44:14,028 INFO   [ForkJoinPool.commonPool-worker-3] CommonMasterDataServiceImpl: Cache: Retrieving Config Data details.我试过下面的链接http://shengwangi.blogspot.com/2015/09/using-log-mdc-in-multi-thread-helloworld-example.html?_sm_au_=iVVrZDSwwf0vP6MR这非常适合TaskExecutor。但我还没有找到 CompletableFuture 的任何解决方案。

Answer 1

我的解决方案主题是（它可以与 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一个. 显然，您不能在该库中保留上下文，但是在您的代码命中应用程序代码后，此方法仍会保留上下文。MDCAwareCompletableFuturegetMDCAwareCompletionStageCompletableFuture

• 在提供 executor 作为参数时，请确保它是 MDC Aware，例如MDCAwareForkJoinPool. 您也可以MDCAwareThreadPoolExecutor通过覆盖execute方法创建以服务于您的用例。你明白了！

您可以在一篇关于相同内容的帖子中找到上述所有内容的详细说明。

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

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

Answer 2

创建包装方法

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.