Wie verwende ich MDC mit Thread-Pools?

Ja, das ist ein häufiges Problem, auf das ich auch gestoßen bin. Es gibt einige Problemumgehungen (wie manuell einstellen, wie beschrieben), aber im Idealfall möchten Sie dies

• Legt den MDC konsistent fest;
• Vermeidet stillschweigende Fehler, bei denen der MDC nicht korrekt ist, aber Sie wissen es nicht; und
• Minimiert Änderungen an der Verwendung von Thread-Pools (z. B. Unterklassifizierung von Callable mit MyCallable überall oder ähnlicher Hässlichkeit).

Hier ist eine Lösung, die ich verwende, um diese drei Anforderungen zu erfüllen. Code sollte selbsterklärend sein.

(Als Randbemerkung kann dieser Executor erstellt und an Guavas MoreExecutors.listeningDecorator() übergeben werden, wenn Sie Guavas ListanableFuture verwenden.)

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);
                    }
                }
            }
        };
    }
}

Wir sind auf ein ähnliches Problem gestoßen. Möglicherweise möchten Sie ThreadPoolExecutor erweitern und die Methoden before/afterExecute überschreiben, um die erforderlichen MDC-Aufrufe auszuführen, bevor Sie neue Threads starten oder stoppen.

IMHO ist die beste Lösung:

• verwenden Sie ThreadPoolTaskExecutor
• eigene TaskDecorator implementieren
• benutze es: executor.setTaskDecorator(new LoggingTaskDecorator());

Der Dekorateur kann so aussehen:

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();
            }
        };
    }

}

Ähnlich wie bei den zuvor veröffentlichten Lösungen können die Methoden newTaskFor für Runnable und Callable überschrieben werden, um das Argument (siehe akzeptierte Lösung) beim Erstellen der RunnableFuture zu umschließen. 

Hinweis: Daher muss die executorService-Methode der Variable submit anstelle der Methode execute aufgerufen werden.

Bei der Variable ScheduledThreadPoolExecutor würden stattdessen die Methoden decorateTask überschrieben.

Dies konnte ich mit folgendem Ansatz lösen

Im Hauptthread (Application.Java, Einstiegspunkt meiner Anwendung)

static public Map<String, String> mdcContextMap = MDC.getCopyOfContextMap();

In der run-Methode der Klasse, die von Executer aufgerufen wird

MDC.setContextMap(Application.mdcContextMap);

So mache ich es mit festen Thread-Pools und Executoren:

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

Im Threading-Teil:

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