如何在线程池中使用MDC？

Question

在我们的软件中,我们广泛使用MDC来跟踪会话ID和Web请求的用户名等内容.这在原始线程中运行时工作正常.但是,有很多事情需要在后台处理.为此,我们使用java.concurrent.ThreadPoolExecutor和java.util.Timer类以及一些自动异步执行服务.所有这些服务都管理自己的线程池.

这就是Logback的手册在这样的环境中使用MDC所要说的:

映射的诊断上下文的副本不能始终由来自启动线程的工作线程继承.当java.util.concurrent.Executors用于线程管理时就是这种情况.例如,newCachedThreadPool方法创建一个ThreadPoolExecutor,就像其他线程池代码一样,它具有复杂的线程创建逻辑.

在这种情况下,建议在将任务提交给执行程序之前,在原始(主)线程上调用MDC.getCopyOfContextMap().当任务运行时,作为其第一个操作,它应调用MDC.setContextMapValues()以将原始MDC值的存储副本与新的Executor托管线程相关联.

这样会很好,但是很容易忘记添加这些调用,并且没有简单的方法来识别问题,直到为时已晚.Log4j的唯一标志是您在日志中丢失了MDC信息,而使用Logback,您会收到过时的MDC信息(因为胎面池中的线程从其上运行的第一个任务继承其MDC).两者都是生产系统中的严重问题.

我不认为我们的情况有任何特殊之处,但我在网上找不到这个问题.显然,这不是很多人碰到的东西,所以必须有办法避免它.我们在这做错了什么？

Answer 1

是的,这是我遇到的一个常见问题.有一些解决方法(如手动设置,如上所述),但理想情况下,您需要一个解决方案

• 一致地设置MDC;
• 避免MDC不正确但您不知道的默认错误; 和
• 最大限度地减少对线程池使用方式的更改(例如Callable,MyCallable在任何地方进行子类化或类似的丑陋).

这是我使用的解决方案,满足这三个需求.代码应该是不言自明的.

(作为旁注,MoreExecutors.listeningDecorator()如果您使用番石榴,可以创建此执行程序并将其提供给Guava ListanableFuture.)

import org.slf4j.MDC;

import java.util.Map;
import java.util.concurrent.*;

/**
 * A SLF4J MDC-compatible {@link ThreadPoolExecutor}.
 * <p/>
 * In general, MDC is used to store diagnostic information (e.g. a user's session id) in per-thread variables, to facilitate
 * logging. However, although MDC data is passed to thread children, this doesn't work when threads are reused in a
 * thread pool. This is a drop-in replacement for {@link ThreadPoolExecutor} sets MDC data before each task appropriately.
 * <p/>
 * Created by jlevy.
 * Date: 6/14/13
 */
public class MdcThreadPoolExecutor extends ThreadPoolExecutor {

    final private boolean useFixedContext;
    final private Map<String, Object> fixedContext;

    /**
     * Pool where task threads take MDC from the submitting thread.
     */
    public static MdcThreadPoolExecutor newWithInheritedMdc(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                                                            TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        return new MdcThreadPoolExecutor(null, corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    /**
     * Pool where task threads take fixed MDC from the thread that creates the pool.
     */
    @SuppressWarnings("unchecked")
    public static MdcThreadPoolExecutor newWithCurrentMdc(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                                                          TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        return new MdcThreadPoolExecutor(MDC.getCopyOfContextMap(), corePoolSize, maximumPoolSize, keepAliveTime, unit,
                workQueue);
    }

    /**
     * Pool where task threads always have a specified, fixed MDC.
     */
    public static MdcThreadPoolExecutor newWithFixedMdc(Map<String, Object> fixedContext, int corePoolSize,
                                                        int maximumPoolSize, long keepAliveTime, TimeUnit unit,
                                                        BlockingQueue<Runnable> workQueue) {
        return new MdcThreadPoolExecutor(fixedContext, corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    private MdcThreadPoolExecutor(Map<String, Object> fixedContext, int corePoolSize, int maximumPoolSize,
                                  long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        this.fixedContext = fixedContext;
        useFixedContext = (fixedContext != null);
    }

    @SuppressWarnings("unchecked")
    private Map<String, Object> getContextForTask() {
        return useFixedContext ? fixedContext : MDC.getCopyOfContextMap();
    }

    /**
     * All executions will have MDC injected. {@code ThreadPoolExecutor}'s submission methods ({@code submit()} etc.)
     * all delegate to this.
     */
    @Override
    public void execute(Runnable command) {
        super.execute(wrap(command, getContextForTask()));
    }

    public static Runnable wrap(final Runnable runnable, final Map<String, Object> context) {
        return new Runnable() {
            @Override
            public void run() {
                Map previous = MDC.getCopyOfContextMap();
                if (context == null) {
                    MDC.clear();
                } else {
                    MDC.setContextMap(context);
                }
                try {
                    runnable.run();
                } finally {
                    if (previous == null) {
                        MDC.clear();
                    } else {
                        MDC.setContextMap(previous);
                    }
                }
            }
        };
    }
}

Answer 2

我们遇到了类似的问题.您可能希望扩展ThreadPoolExecutor并覆盖before/afterExecute方法,以在启动/停止新线程之前进行所需的MDC调用.

Answer 3

恕我直言,最好的解决方案是:

• 使用 ThreadPoolTaskExecutor
• 实现自己的 TaskDecorator
• 用它: executor.setTaskDecorator(new LoggingTaskDecorator());

装饰者看起来像这样:

private final class LoggingTaskDecorator implements TaskDecorator {

    @Override
    public Runnable decorate(Runnable task) {
        // web thread
        Map<String, String> webThreadContext = MDC.getCopyOfContextMap();
        return () -> {
            // work thread
            try {
                // TODO: is this thread safe?
                MDC.setContextMap(webThreadContext);
                task.run();
            } finally {
                MDC.clear();
            }
        };
    }

}

Answer 4

这就是我使用固定线程池和执行程序的方式：

ExecutorService executor = Executors.newFixedThreadPool(4);
Map<String, String> mdcContextMap = MDC.getCopyOfContextMap();

在穿线部分：

executor.submit(() -> {
    MDC.setContextMap(mdcContextMap);
    // my stuff
});

Answer 5

如果您在使用@Async注解运行任务的 spring 框架相关环境中遇到此问题，您可以使用TaskDecorator方法来装饰任务。

此处提供了如何执行此操作的示例：

• Spring 4.3：使用 TaskDecorator 将 MDC 数据复制到 @Async 线程

我遇到了这个问题，上面的文章帮助我解决了这个问题，这就是我在这里分享的原因。