SpringBoot异步线程间传递上下文
 
需求
实现
启用异步功能
配置异步
配置任务装饰器
完工
 

 
需求
 
SpringBoot项目中，经常使用@Async来开启一个子线程来完成异步操作。主线程中的用户信息需要传递给子线程
 
实现
 
启用异步功能
 
在启动类里加上@EnableAsync注解
 
@EnableAsync
@SpringBootApplication
public class Application {}
 
配置异步
 
新建一个配置类，实现AsyncConfigurer接口，并重写getAsyncExecutor方法
 
@Configuration
public class AsyncConfig implements AsyncConfigurer {

    @Override
    public Executor getAsyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(50);
        executor.setThreadNamePrefix("async-pool-");
        // 这一步是关键，异步Task装饰器
        executor.setTaskDecorator(new MyContextDecorator());
        executor.initialize();
        return executor;
    }
}
 
配置任务装饰器
 
新建一个异步任务装饰器，实现TaskDecorator接口，并重写decorate方法
 
public class MyContextDecorator implements TaskDecorator {
    @Override
    @Nonnull
    public Runnable decorate(@Nonnull Runnable runnable) {

		// 获取主线程中的请求信息（我们的用户信息也放在里面）
       RequestAttributes attributes = RequestContextHolder.getRequestAttributes();
        return () -> {
            try {
              	// 将主线程的请求信息，设置到子线程中
              	RequestContextHolder.setRequestAttributes(attributes);
             	// 执行子线程，这一步不要忘了
                runnable.run();
            } finally {
            	// 线程结束，清空这些信息，否则可能造成内存泄漏
                RequestContextHolder.resetRequestAttributes();
            }
        };
    }
 
补充：RequestContextHolder内部是基于ThreadLocal实现的，因此在使用set get时，都是和当前线程绑定的。当然，使用者的用户信息不一定放在了RequestContextHolder里面，读者可以自行扩展。
 
完工
 
到此，通过@Async开启的子线程，就可以正常拿到父线程中的Request信息了。

后续文章首发在个人博客，欢迎移驾我的个人博客浏览该文章，地址在下方
 
多线程篇-父子线程的上下文传递

该项目已合并到 git项目microservice-sc-v1.0 dev分支中，具体位置为microservice-sc-v1.0/module-service-hi/src/main/java/servicehi/asyncthreadaop/，假如对您的需求有帮助，可不可以不吝给颗star，您的肯定是我的动力。
 
从朋友那里借鉴来的思路，他当时遇到的业务情况是这样的：
 
每个用户登陆系统后，该用户需要异步执行多个方法，方法内涉及到从securityContext和LogContext中读取用户信息
 
抽象出来便是
 
用户登录的主线程需要异步(多子线程)执行多个方法，方法中需要保持用户的上下文信息
 
解决思路：
 
aop拦截子线程的调用，将上下文通过封装的Runnable对象传递给子线程。因为aop是beanPostProcessor后处理器，只能拦截1)bean中的2)方法，所以逆推得 子线程来自 一个2)线程池的submit执行合适一些，而且这个线程池对象是1)一个bean。
 
实现：
 
1、bean形式存在的线程池
 
@Configuration
public class AsynExecutor {
	
	@Bean
	@Lazy
	public ExecutorService defaultExecutor() {
		return Executors.newCachedThreadPool();
	}

}
 
2、该线程池提供的线程可以通过注解的方式被业务方法体异步使用
 
@Async("defaultExecutor")
@Target({ElementType.METHOD,ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface DefaultAsyncThread {

	
}
 
3、标注如上注解的方法在被线程池中线程执行时添加AOP
 
public class DefaultExecutorSubmitAspect {

	@Around("execution(* java.util.concurrent.Executor.*(..))")
	@SneakyThrows
	public Object intercept(ProceedingJoinPoint joinPoint) {
		
		Object[] args = joinPoint.getArgs();
		for(int i =0; i< args.length; i++) {
			args[i] = processArgs(args[i]);
		}
		
		return joinPoint.proceed();
	}
	
	Object processArgs(Object arg) {
		if(arg instanceof Runnable) {
			Runnable r = (Runnable)arg;
			return new RunnableWeave(r);
		}
		if(arg instanceof Callable<?>) {
			Callable<?> c = (Callable<?>)arg;
			return new CallableWeave<>(c);
		}
		if(arg instanceof Collection<?>) {
			Collection<?> cs = (Collection<?>)arg;
			List<Object> collect = cs.stream().map(this::processArgs).collect(Collectors.toList());
			return collect;
		}
		return arg;
	}
}
 
4、将上下文传递给子线程，从而保证 子线程里可以读取用户登录信息(父线程上下文信息)
 
public class RunnableWeave  implements Runnable {

	private final Runnable r;
	private final Map<String, String> MDCContextMap;
	private final SecurityContext securityContext;
	
	public RunnableWeave(Runnable r) {
		this.r = r;
		this.MDCContextMap = MDC.getCopyOfContextMap();
		securityContext = SecurityContextHolder.getContext();
	}
	
	@Override
	public void run() {
		MDC.setContextMap(MDCContextMap);
		SecurityContextHolder.setContext(securityContext);
		r.run();
		SecurityContextHolder.clearContext();
		MDC.setContextMap(new HashMap<>());
	}

}
 
public class CallableWeave<V> implements Callable<V> {

	private final Callable<V> c;
	private final Map<String, String> MDCContextMap;
	private final SecurityContext securityContext;
	
	public CallableWeave(Callable<V> c) {
		this.c = c;
		this.MDCContextMap = MDC.getCopyOfContextMap();
		this.securityContext = SecurityContextHolder.getContext();
	}
	@Override
	public V call() throws Exception {
		MDC.setContextMap(MDCContextMap);
		SecurityContextHolder.setContext(securityContext);
		V v = c.call();
		MDC.setContextMap(new HashMap<>());
		SecurityContextHolder.clearContext();
		return v;
	}

}

 
 
一般同步编程模型中我们使用ThreadLocal即可，但是在异步编程模型中（可能有同学有疑问，为什么不用InheritThreadLocal？看这篇记一次线上采坑实录）会导致上下文失效。
 
 
但是spring 4.3给出了好的方案，利用TaskDecorator。
 
看这个名称大概就能猜出是一个装饰器设计原理
 
我们分析下线程池的源码
 
@Override
	protected ExecutorService initializeExecutor(
			ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {

		BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);

		ThreadPoolExecutor executor;
        //当线程池的装饰器不为空时，执行execute方法会进入这里，因为它重写了execute方法
		if (this.taskDecorator != null) {
			executor = new ThreadPoolExecutor(
					this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
					queue, threadFactory, rejectedExecutionHandler) {
                
                //这里是一个代理设计模式的实现，对execute做了一层代理
				@Override
				public void execute(Runnable command) {
                    //执行装饰器的逻辑，注意这段代码是在主线程中运行
					Runnable decorated = taskDecorator.decorate(command);
					if (decorated != command) {
						decoratedTaskMap.put(decorated, command);
					}
                    //子线程真正执行的方法（异步模块）...初始化核心线程数，核心线程满了入队列，队列满开启至最大线程数
					super.execute(decorated);
				}
			};
		}
		else {
			executor = new ThreadPoolExecutor(
					this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
					queue, threadFactory, rejectedExecutionHandler);

		}

		if (this.allowCoreThreadTimeOut) {
			executor.allowCoreThreadTimeOut(true);
		}

		this.threadPoolExecutor = executor;
		return executor;
	}
 
demo演示
 
主线程16个，子线程2个，执行10次，目的是尽可能让子线程复用。
 
@Test
    public void testThreadLocal() {
        ThreadLocal<Integer> threadLocal = new ThreadLocal<>();
        ExecutorService mainThreadPool = Executors.newFixedThreadPool(16);

        ThreadPoolTaskExecutor childThreadPool = new ThreadPoolTaskExecutor();
        childThreadPool.setCorePoolSize(2);
        childThreadPool.setMaxPoolSize(2);

        childThreadPool.setTaskDecorator(runnable -> {
            int v = threadLocal.get();
            System.out.println("装饰器中获取到主线程=" + Thread.currentThread().getName() + " 获取上下文=" + v);
            return () -> {
                try {
                    //重新copy传递给子线程
                    threadLocal.set(v);
                    runnable.run();
                } finally {
                    threadLocal.remove();
                }
            };
        });
        childThreadPool.initialize();

        for (int i = 0; i < 10; i++) {
            int finalI = i;
            mainThreadPool.execute(() -> {
                //模拟在主线程设置上下文变量
                threadLocal.set(finalI);
                childThreadPool.execute(() -> System.out.println("子线程" + Thread.currentThread().getName() + " 获取上下文变量=" + threadLocal.get()));
                threadLocal.remove();
            });
        }
        try {
            childThreadPool.getThreadPoolExecutor().awaitTermination(3, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
 
 可以看到，我们传递的值都能正确获取到，没有像inheritThreadLocal出现紊乱。
 
2020-12-03 10:59:26.350 [main] INFO  o.s.s.c.ThreadPoolTaskExecutor - Initializing ExecutorService
装饰器中获取到主线程=pool-1-thread-1 获取上下文=0
装饰器中获取到主线程=pool-1-thread-2 获取上下文=1
装饰器中获取到主线程=pool-1-thread-3 获取上下文=2
装饰器中获取到主线程=pool-1-thread-4 获取上下文=3
装饰器中获取到主线程=pool-1-thread-5 获取上下文=4
装饰器中获取到主线程=pool-1-thread-6 获取上下文=5
装饰器中获取到主线程=pool-1-thread-8 获取上下文=7
装饰器中获取到主线程=pool-1-thread-10 获取上下文=9
装饰器中获取到主线程=pool-1-thread-9 获取上下文=8
装饰器中获取到主线程=pool-1-thread-7 获取上下文=6

Disconnected from the target VM, address: '127.0.0.1:4991', transport: 'socket'
子线程ThreadPoolTaskExecutor-1 获取上下文变量=2
子线程ThreadPoolTaskExecutor-1 获取上下文变量=3
子线程ThreadPoolTaskExecutor-1 获取上下文变量=1
子线程ThreadPoolTaskExecutor-1 获取上下文变量=0
子线程ThreadPoolTaskExecutor-1 获取上下文变量=5
子线程ThreadPoolTaskExecutor-1 获取上下文变量=7
子线程ThreadPoolTaskExecutor-1 获取上下文变量=9
子线程ThreadPoolTaskExecutor-1 获取上下文变量=8
子线程ThreadPoolTaskExecutor-1 获取上下文变量=6
子线程ThreadPoolTaskExecutor-2 获取上下文变量=4

Process finished with exit code 0