使用 @Async 和 TaskDecorator 记录 MDC

Question

使用 Spring MVC，我有以下设置：

1. 用于记录请求的 AbstractRequestLoggingFilter 派生过滤器。
2. TaskDecorator 用于编组从 Web 请求线程到 @Async 线程的 MDC 上下文映射。

我正在尝试使用 MDC（或 ThreadLocal 对象）为处理请求所涉及的所有组件收集上下文信息。

我可以从@Async 线程正确检索MDC 上下文信息。但是，如果 @Async 线程要将上下文信息添加到 MDC，我现在如何将 MDC 上下文信息编组到处理响应的线程？

任务装饰器

public class MdcTaskDecorator implements TaskDecorator {
@Override
public Runnable decorate(Runnable runnable) {
    // Web thread context
    // Get the logging MDC context
    Map<String, String> contextMap = MDC.getCopyOfContextMap();

    return () -> {
        try {
            // @Async thread context
            // Restore the web thread MDC context
            if(contextMap != null) {
                MDC.setContextMap(contextMap);
            }
            else {
                MDC.clear();
            }

            // Run the new thread
            runnable.run();
        }
        finally {
            MDC.clear();
        }
    };
}

}

异步方法

@Async
public CompletableFuture<String> doSomething_Async() {
    MDC.put("doSomething", "started");
    return doit();
}

日志过滤器

public class ServletLoggingFilter extends AbstractRequestLoggingFilter {
@Override
protected void beforeRequest(HttpServletRequest request, String message) {
    MDC.put("webthread", Thread.currentThread().getName()); // Will be webthread-1
}

@Override
protected void afterRequest(HttpServletRequest request, String message) {
    MDC.put("responsethread", Thread.currentThread().getName()); // Will be webthread-2
    String s = MDC.get("doSomething"); // Will be null

    // logthis();
}

}

Answer 1

我希望你已经解决了这个问题，但如果你没有，这里有一个解决方案。
您所要做的可以总结为以下两个简单的步骤：

1. 继续上课MdcTaskDecorator。
2. 为您的主类扩展 AsyncConfigurerSupport 并覆盖 getAsyncExecutor() 以使用您自定义的装饰器设置如下：

    public class AsyncTaskDecoratorApplication extends AsyncConfigurerSupport {
        @Override
        public Executor getAsyncExecutor() {
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setTaskDecorator(new MdcTaskDecorator());
            executor.initialize();
            return executor;
        }

        public static void main(String[] args) {
            SpringApplication.run(AsyncTaskdecoratorApplication.class, args);
        }
    }

Answer 2

创建一个将 MDC 属性从父线程传递到后继线程的 bean。

@Configuration
@Slf4j
public class AsyncMDCConfiguration {
  @Bean
  public Executor asyncExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setTaskDecorator(new MDCTaskDecorator());//MDCTaskDecorator i s a custom created  class thet implements  TaskDecorator  that is reponsible for passing on the MDC properties 
    executor.initialize();
    return executor;
  }
}


@Slf4j
public class MDCTaskDecorator implements TaskDecorator {

  @Override
  public Runnable decorate(Runnable runnable) {
    Map<String, String> contextMap = MDC.getCopyOfContextMap();
    return () -> {
      try {
        MDC.setContextMap(contextMap);
        runnable.run();
      } finally {
        MDC.clear();
      }
    };
  }
}

现在一切都好。快乐编码

Answer 3

我有一些解决方案，大致分为Callable(for @Async)、AsyncExecutionInterceptor(for @Async)、CallableProcessingInterceptor(for controller)。

1.将上下文信息放入@Async线程的Callable解决方案：

关键是使用ContextAwarePoolExecutor来替换@Async的默认执行器：

@Configuration

公共类 DemoExecutorConfig {

@Bean("demoExecutor")
public Executor contextAwarePoolExecutor() {
    return new ContextAwarePoolExecutor();
}

}

而 ContextAwarePoolExecutor 使用 ContextAwareCallable 覆盖 submit 和 submitListenable 方法：

public class ContextAwarePoolExecutor extends ThreadPoolTaskExecutor {

private static final long serialVersionUID = 667815067287186086L;

@Override
public <T> Future<T> submit(Callable<T> task) {
    return super.submit(new ContextAwareCallable<T>(task, newThreadContextContainer()));
}

@Override
public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
    return super.submitListenable(new ContextAwareCallable<T>(task, newThreadContextContainer()));
}

/**
 * set infos what we need
 */
private ThreadContextContainer newThreadContextContainer() {
    ThreadContextContainer container = new ThreadContextContainer();
    container.setRequestAttributes(RequestContextHolder.currentRequestAttributes());
    container.setContextMapOfMDC(MDC.getCopyOfContextMap());
    return container;
}

}

ThreadContextContainer 只是一个方便存储信息的 pojo：

public class ThreadContextContainer implements Serializable {

private static final long serialVersionUID = -6809291915300091330L;

private RequestAttributes requestAttributes;
private Map<String, String> contextMapOfMDC;

public RequestAttributes getRequestAttributes() {
    return requestAttributes;
}

public Map<String, String> getContextMapOfMDC() {
    return contextMapOfMDC;
}

public void setRequestAttributes(RequestAttributes requestAttributes) {
    this.requestAttributes = requestAttributes;
}

public void setContextMapOfMDC(Map<String, String> contextMapOfMDC) {
    this.contextMapOfMDC = contextMapOfMDC;
}

}

ContextAwareCallable（原始任务的可调用代理）在原始任务执行其调用方法之前覆盖调用方法以存储 MDC 或其他上下文信息：

public class ContextAwareCallable<T> implements Callable<T> {

/**
 * the original task
 */
private Callable<T> task;

/**
 * for storing infos what we need
 */
private ThreadContextContainer threadContextContainer;

public ContextAwareCallable(Callable<T> task, ThreadContextContainer threadContextContainer) {
    this.task = task;
    this.threadContextContainer = threadContextContainer;
}

@Override
public T call() throws Exception {
    // set infos
    if (threadContextContainer != null) {
        RequestAttributes requestAttributes = threadContextContainer.getRequestAttributes();
        if (requestAttributes != null) {
            RequestContextHolder.setRequestAttributes(requestAttributes);
        }
        Map<String, String> contextMapOfMDC = threadContextContainer.getContextMapOfMDC();
        if (contextMapOfMDC != null) {
            MDC.setContextMap(contextMapOfMDC);
        }
    }

    try {
        // execute the original task
        return task.call();
    } finally {
        // clear infos after task completed
        RequestContextHolder.resetRequestAttributes();
        try {
            MDC.clear();
        } finally {
        }
    }
}

}

最后，像这样使用@Async 和配置的bean“demoExecutor”：@Async("demoExecutor") void yourTaskMethod();

2.关于您处理回复的问题：

遗憾地告诉我，我真的没有经过验证的解决方案。也许 org.springframework.aop.interceptor.AsyncExecutionInterceptor#invoke 可以解决这个问题。

而且我认为它没有解决方案来处理您的 ServletLoggingFilter 的响应。因为 Async 方法会立即返回。 afterRequest 方法立即执行并在 Async 方法执行操作之前返回。除非您同步等待 Async 方法完成执行，否则您不会得到您想要的。

但如果你只是想记录一些东西，你可以在原始任务执行其调用方法之后将这些代码添加到我的示例 ContextAwareCallable 中：

try {
        // execute the original task
        return task.call();
    } finally {
        String something = MDC.get("doSomething"); // will not be null
        // logthis(something);

        // clear infos after task completed
        RequestContextHolder.resetRequestAttributes();
        try {
            MDC.clear();
        } finally {
        }
    }