java - ejemplo - ¿Cómo usar MDC con pools de hilos?

executorservice java ejemplo

Question

En nuestro software, utilizamos extensamente MDC para rastrear cosas como ID de sesión y nombres de usuario para solicitudes web. Esto funciona bien mientras se ejecuta en el hilo original. Sin embargo, hay muchas cosas que deben procesarse en segundo plano. Para eso usamos las clases java.concurrent.ThreadPoolExecutor y java.util.Timer junto con algunos servicios de ejecución asincrónicos automáticos. Todos estos servicios administran su propio grupo de subprocesos.

Esto es lo que el manual de Logback tiene que decir sobre el uso de MDC en dicho entorno:

Una copia del contexto de diagnóstico mapeado no siempre puede ser heredada por subprocesos de trabajo del subproceso iniciador. Este es el caso cuando java.util.concurrent.Executors se usa para la gestión de subprocesos. Por ejemplo, el método newCachedThreadPool crea un ThreadPoolExecutor y, al igual que otro código de agrupación de subprocesos, tiene una intrincada lógica de creación de subprocesos.

En tales casos, se recomienda invocar a MDC.getCopyOfContextMap () en el hilo original (principal) antes de enviar una tarea al ejecutor. Cuando la tarea se ejecuta, como primera acción, debe invocar a MDC.setContextMapValues ​​() para asociar la copia almacenada de los valores originales de MDC con el nuevo hilo administrado por Executor.

Esto estaría bien, pero es muy fácil olvidar agregar esas llamadas, y no hay una manera fácil de reconocer el problema hasta que sea demasiado tarde. El único signo con Log4j es que se pierde información MDC en los registros, y con Logback obtienes información MDC obsoleta (ya que el subproceso en el grupo de la banda de rodadura hereda su MDC de la primera tarea que se ejecutó en él). Ambos son problemas serios en un sistema de producción.

No veo nuestra situación especial de ninguna manera, sin embargo, no pude encontrar mucho sobre este problema en la web. Aparentemente, esto no es algo contra lo que mucha gente se enfrenta, así que debe haber una forma de evitarlo. ¿Qué estamos haciendo mal aquí?

Answer 1

Así es como lo hago con grupos de hilos fijos y ejecutores:

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

En la parte de enhebrado:

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

Answer 2

De forma similar a las soluciones publicadas anteriormente, los métodos newTaskFor para Runnable y Callable se pueden sobrescribir para envolver el argumento (ver solución aceptada) al crear RunnableFuture .

Nota: En consecuencia, se debe submit método de submit del executorService lugar del método de execute .

Para el ScheduledThreadPoolExecutor , los métodos decorateTask se sobrescribirán en su lugar.

Answer 3

En mi humilde opinión, la mejor solución es:

• usa ThreadPoolTaskExecutor
• implementa tu propio TaskDecorator
• executor.setTaskDecorator(new LoggingTaskDecorator()); : executor.setTaskDecorator(new LoggingTaskDecorator());

El decorador puede verse así:

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

Nos encontramos con un problema similar. Es posible que desee extender ThreadPoolExecutor y anular los métodos berfore / afterExecute para hacer las llamadas MDC que necesita antes de iniciar / detener nuevos subprocesos.

Answer 5

Pude resolver esto utilizando el siguiente enfoque

En el hilo principal (Application.java, el punto de entrada de mi aplicación)

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

En el método de ejecución de la clase que recibe el nombre de Executer

MDC.setContextMap(Application.mdcContextMap);

Answer 6

Sí, este es un problema común que me he encontrado también. Hay algunas soluciones (como configurarlo manualmente, como se describe), pero lo ideal es que quiera una solución que

• Establece el MDC consistentemente;
• Evita errores tácitos donde el MDC es incorrecto pero no lo sabes; y
• Minimiza los cambios en la forma en que usa los grupos de subprocesos (por ejemplo, subclases de Callable con MyCallable todas partes, o similar fealdad).

Aquí hay una solución que uso que cumple con estas tres necesidades. El código debe ser autoexplicativo.

(Como nota al margen, este ejecutor se puede crear y alimentar a MoreExecutors.listeningDecorator() Guava, si usa 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);
}
}
}
};
}
}