Como usar o MDC com conjuntos de threads?

Em nosso software, usamos extensivamente o MDC para rastrear itens como IDs de sessão e nomes de usuário para solicitações da Web. Isso funciona bem durante a execução no encadeamento original. No entanto, há muitas coisas que precisam ser processadas em segundo plano. Para isso, usamos as classes java.concurrent.ThreadPoolExecutore java.util.Timerjuntamente com alguns serviços de execução assíncrona autolaminados. Todos esses serviços gerenciam seu próprio pool de encadeamentos.

Isto é o que o manual do Logback tem a dizer sobre o uso do MDC em um ambiente como esse:

Uma cópia do contexto de diagnóstico mapeado nem sempre pode ser herdada pelos threads de trabalho do thread inicial. Este é o caso em que java.util.concurrent.Executors é usado para gerenciamento de encadeamentos. Por exemplo, o método newCachedThreadPool cria um ThreadPoolExecutor e, como outro código de pool de encadeamentos, possui uma lógica intrincada de criação de encadeamentos.

Nesses casos, é recomendável que MDC.getCopyOfContextMap () seja chamado no encadeamento original (mestre) antes de enviar uma tarefa ao executor. Quando a tarefa é executada, como primeira ação, ela deve chamar MDC.setContextMapValues ​​() para associar a cópia armazenada dos valores originais do MDC ao novo encadeamento gerenciado pelo Executor.

Isso seria bom, mas é muito fácil esquecer de adicionar essas chamadas e não há uma maneira fácil de reconhecer o problema até que seja tarde demais. O único sinal com o Log4j é que você obtém informações ausentes do MDC nos logs e, com o Logback, obtém informações obsoletas do MDC (uma vez que o encadeamento no pool do piso herda o MDC da primeira tarefa executada nele). Ambos são problemas sérios em um sistema de produção.

Não vejo nossa situação de forma alguma, mas não pude encontrar muito sobre esse problema na web. Aparentemente, isso não é algo contra o qual muitas pessoas se deparam, então deve haver uma maneira de evitá-lo. O que estamos fazendo de errado aqui?

Sim, esse também é um problema comum. Existem algumas soluções alternativas (como configurá-lo manualmente, conforme descrito), mas, idealmente, você deseja uma solução que

• Define o MDC de forma consistente;
• Evita erros tácitos onde o MDC está incorreto, mas você não o conhece; e
• Minimiza as alterações na maneira como você usa conjuntos de encadeamentos (por exemplo, subclassificação Callableem MyCallabletodos os lugares ou feiúra semelhante).

Aqui está uma solução que eu uso que atende a essas três necessidades. O código deve ser auto-explicativo.

(Como uma observação lateral, esse executor pode ser criado e alimentado no Guava's MoreExecutors.listeningDecorator(), se você usar o Guava's 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);
                    }
                }
            }
        };
    }
}

Nós encontramos um problema semelhante. Convém estender o ThreadPoolExecutor e substituir os métodos before / afterExecute para fazer as chamadas MDC necessárias antes de iniciar / parar novos threads.

IMHO a melhor solução é:

• usar ThreadPoolTaskExecutor
• implemente seu próprio TaskDecorator
• use-o: executor.setTaskDecorator(new LoggingTaskDecorator());

O decorador pode ficar assim:

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

}

É assim que eu faço com conjuntos de threads e executores fixos:

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

Na parte de rosqueamento:

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

Semelhante às soluções postadas anteriormente, os newTaskFormétodos Runnablee Callablepodem ser substituídos para envolver o argumento (consulte a solução aceita) ao criar o arquivo RunnableFuture.

Nota: Por conseguinte, o executorService's submitmétodo deve ser chamado em vez do executemétodo.

Para o ScheduledThreadPoolExecutor, os decorateTaskmétodos seriam substituídos.

Caso você enfrente esse problema em um ambiente relacionado à estrutura de primavera em que executa tarefas usando a @Asyncanotação, é possível decorar as tarefas usando a abordagem TaskDecorator . Uma amostra de como fazer isso é fornecida aqui: https://moelholm.com/blog/2017/07/24/spring-43-using-a-taskdecorator-to-copy-mdc-data-to-async-threads

Eu enfrentei esse problema e o artigo acima me ajudou a resolvê-lo, por isso estou compartilhando aqui.

Outra variação semelhante às respostas existentes aqui é implementar ExecutorServicee permitir que um delegado seja passado a ele. Em seguida, usando genéricos, ele ainda pode expor o delegado real, caso deseje obter algumas estatísticas (desde que nenhum outro método de modificação seja usado).

Código de referência:

• https://github.com/project-ncl/pnc/blob/master/common/src/main/java/org/jboss/pnc/common/concurrent/MDCThreadPoolExecutor.java
• https://github.com/project-ncl/pnc/blob/master/common/src/main/java/org/jboss/pnc/common/concurrent/MDCWrappers.java

public class MDCExecutorService<D extends ExecutorService> implements ExecutorService {

    private final D delegate;

    public MDCExecutorService(D delegate) {
        this.delegate = delegate;
    }

    @Override
    public void shutdown() {
        delegate.shutdown();
    }

    @Override
    public List<Runnable> shutdownNow() {
        return delegate.shutdownNow();
    }

    @Override
    public boolean isShutdown() {
        return delegate.isShutdown();
    }

    @Override
    public boolean isTerminated() {
        return delegate.isTerminated();
    }

    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        return delegate.awaitTermination(timeout, unit);
    }

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        return delegate.submit(wrap(task));
    }

    @Override
    public <T> Future<T> submit(Runnable task, T result) {
        return delegate.submit(wrap(task), result);
    }

    @Override
    public Future<?> submit(Runnable task) {
        return delegate.submit(wrap(task));
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
        return delegate.invokeAll(wrapCollection(tasks));
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException {
        return delegate.invokeAll(wrapCollection(tasks), timeout, unit);
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
        return delegate.invokeAny(wrapCollection(tasks));
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        return delegate.invokeAny(wrapCollection(tasks), timeout, unit);
    }

    @Override
    public void execute(Runnable command) {
        delegate.execute(wrap(command));
    }

    public D getDelegate() {
        return delegate;
    }

    /* Copied from https://github.com/project-ncl/pnc/blob/master/common/src/main/java/org/jboss/pnc/common
    /concurrent/MDCWrappers.java */

    private static Runnable wrap(final Runnable runnable) {
        final Map<String, String> context = MDC.getCopyOfContextMap();
        return () -> {
            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);
                }
            }
        };
    }

    private static <T> Callable<T> wrap(final Callable<T> callable) {
        final Map<String, String> context = MDC.getCopyOfContextMap();
        return () -> {
            Map previous = MDC.getCopyOfContextMap();
            if (context == null) {
                MDC.clear();
            } else {
                MDC.setContextMap(context);
            }
            try {
                return callable.call();
            } finally {
                if (previous == null) {
                    MDC.clear();
                } else {
                    MDC.setContextMap(previous);
                }
            }
        };
    }

    private static <T> Consumer<T> wrap(final Consumer<T> consumer) {
        final Map<String, String> context = MDC.getCopyOfContextMap();
        return (t) -> {
            Map previous = MDC.getCopyOfContextMap();
            if (context == null) {
                MDC.clear();
            } else {
                MDC.setContextMap(context);
            }
            try {
                consumer.accept(t);
            } finally {
                if (previous == null) {
                    MDC.clear();
                } else {
                    MDC.setContextMap(previous);
                }
            }
        };
    }

    private static <T> Collection<Callable<T>> wrapCollection(Collection<? extends Callable<T>> tasks) {
        Collection<Callable<T>> wrapped = new ArrayList<>();
        for (Callable<T> task : tasks) {
            wrapped.add(wrap(task));
        }
        return wrapped;
    }
}

Consegui resolver isso usando a seguinte abordagem

No thread principal (Application.java, o ponto de entrada do meu aplicativo)

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

No método de execução da classe que é chamada pelo Executer

MDC.setContextMap(Application.mdcContextMap);