多线程调用如何传递请求上下文？简述ThreadLocal和TaskDecorator
Posted OkidoGreen
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了多线程调用如何传递请求上下文？简述ThreadLocal和TaskDecorator相关的知识，希望对你有一定的参考价值。
https://www.jianshu.com/p/4ec1f0b993cd背景
微服务应用大多对外提供RESTful API，需要有相应的token才能访问，我们在聚合服务中使用Feign Client调用这些API，顺序执行往往会浪费大量的IO等待时间，为了提高查询速度，我们会使用异步调用，Java 8引入了CompletableFuture，结合Executor框架大大简化了异步编程的复杂性。
问题描述
我们的服务使用Spring Security OAuth2授权，并通过JWT传递token，对于用户侧的请求一律使用用户token，后端服务间的调用使用系统token。但有时为了简化开发（如用户信息传递），服务间调用仍然传递用户token。这就带来一个问题，在异步请求中子线程会丢失相应的认证信息，错误如下：
org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'scopedTarget.oauth2ClientContext': Scope 'request' is not active for the current thread; consider defining a scoped proxy for this bean if you intend to refer to it from a singleton;
at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:365)
at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:199)
at org.springframework.aop.target.SimpleBeanTargetSource.getTarget(SimpleBeanTargetSource.java:35)
at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:192)
at com.sun.proxy.$Proxy170.getAccessToken(Unknown Source)
at org.springframework.cloud.security.oauth2.client.feign.OAuth2FeignRequestInterceptor.getToken(OAuth2FeignRequestInterceptor.java:126)
at org.springframework.cloud.security.oauth2.client.feign.OAuth2FeignRequestInterceptor.extract(OAuth2FeignRequestInterceptor.java:115)
at org.springframework.cloud.security.oauth2.client.feign.OAuth2FeignRequestInterceptor.apply(OAuth2FeignRequestInterceptor.java:104)
at feign.SynchronousMethodHandler.targetRequest(SynchronousMethodHandler.java:169)
at feign.SynchronousMethodHandler.executeAndDecode(SynchronousMethodHandler.java:99)
at feign.SynchronousMethodHandler.invoke(SynchronousMethodHandler.java:78)
...
Caused by: java.lang.IllegalStateException: No thread-bound request found: Are you referring to request attributes outside of an actual web request, or processing a request outside of the originally receiving thread? If you are actually operating within a web request and still receive this message, your code is probably running outside of DispatcherServlet: In this case, use RequestContextListener or RequestContextFilter to expose the current request.
at org.springframework.web.context.request.RequestContextHolder.currentRequestAttributes(RequestContextHolder.java:131)
at org.springframework.web.context.request.AbstractRequestAttributesScope.get(AbstractRequestAttributesScope.java:42)
at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:353)
... 76 common frames omitted
问题暴露出来了，该怎么解决呢？在这之前我们先了解两个概念，线程的ThreadLocal和TaskDecorator。
ThreadLocal<T>的原理
首先，在Thread类中定义了一个threadLocals，它是ThreadLocal.ThreadLocalMap对象的引用,默认值是null。ThreadLocal.ThreadLocalMap对象表示了一个以开放地址形式的散列表。当我们在线程的run方法中第一次调用ThreadLocal对象的get方法时，会为当前线程创建一个ThreadLocalMap对象。也就是每个线程都各自有一张独立的散列表，以ThreadLocal对象作为散列表的key，set方法中的值作为value（第一次调用get方法时，以initialValue方法的返回值作为value）。显然我们可以定义多个ThreadLocal对象，而我们一般将ThreadLocal对象定义为static类型或者外部类中。上面所表达的意思就是，相同的key在不同的散列表中的值必然是独立的，每个线程都是在各自的散列表中执行操作。
ThreadLocal存储
TaskDecorator简介
先看官方api文档说明
public interface TaskDecorator
A callback interface for a decorator to be applied to any Runnable about to be executed.
Note that such a decorator is not necessarily being applied to the user-supplied Runnable/Callable but rather to the actual execution callback (which may be a wrapper around the user-supplied task).
The primary use case is to set some execution context around the task's invocation, or to provide some monitoring/statistics for task execution.
意思就是说这是一个执行回调方法的装饰器，主要应用于传递上下文，或者提供任务的监控/统计信息。看上去正好可以应用于我们这种场景。
解决方案
上文中的错误信息涉及到RequestAttributes
和SecurityContext，他们都是通过ThreadLocal来保存线程数据，在同步方法中没有问题，使用线程池异步调用时，我们可以通过配合线程池的TaskDecorator装饰器拷贝上下文传递。
注意 线程池中的线程是可复用的，使用ThreadLocal需要注意内存泄露问题，所以线程执行完成后需要在finally方法中移除上下文对象。
代码如下
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskDecorator;
import org.springframework.scheduling.annotation.AsyncConfigurer;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.security.core.context.SecurityContext;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.web.context.request.RequestAttributes;
import org.springframework.web.context.request.RequestContextHolder;
import javax.annotation.Nonnull;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer
@Bean("ttlExecutor")
public Executor taskExecutor()
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
// 设置线程池核心容量
executor.setCorePoolSize(20);
// 设置线程池最大容量
executor.setMaxPoolSize(100);
// 设置任务队列长度
executor.setQueueCapacity(200);
// 设置线程超时时间
executor.setKeepAliveSeconds(60);
// 设置线程名称前缀
executor.setThreadNamePrefix("ttl-executor-");
// 设置任务丢弃后的处理策略
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
// 设置任务的装饰
executor.setTaskDecorator(new ContextCopyingDecorator());
executor.initialize();
return executor;
static class ContextCopyingDecorator implements TaskDecorator
@Nonnull
@Override
public Runnable decorate(@Nonnull Runnable runnable)
RequestAttributes context = RequestContextHolder.currentRequestAttributes();
SecurityContext securityContext = SecurityContextHolder.getContext();
return () ->
try
RequestContextHolder.setRequestAttributes(context);
SecurityContextHolder.setContext(securityContext);
runnable.run();
finally
SecurityContextHolder.clearContext();
RequestContextHolder.resetRequestAttributes();
;
扩展知识
Spring安全策略可见性分为三个层级：
- MODE_THREADLOCAL 仅当前线程（默认）
- MODE_INHERITABLETHREADLOCAL 子线程可见
- MODE_GLOBAL 全局可见
可通过启动项参数进行设置
-Dspring.security.strategy=MODE_INHERITABLETHREADLOCAL
参考资料
- RequestContextHolder分析
- alibaba/transmittable-thread-local
- stackoverflow/how-to-enable-request-scope-in-async-task-executor
- 线程池中使用ThreadLocal方案
- ThreadLocal原理及使用示例
ThreadPoolTaskExecutor在执行线程时，存在一个
TaskDecorator配置，可以装饰线程类。
1. 源码分析
源码：org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor#initializeExecutor
@Override protected ExecutorService initializeExecutor(ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler)
BlockingQueue < Runnable > queue = createQueue(this.queueCapacity);
ThreadPoolExecutor executor;
//配置了线程装饰器
if (this.taskDecorator != null)
executor = new ThreadPoolExecutor(this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS, queue, threadFactory, rejectedExecutionHandler) @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;
2. 实际使用
声明spring线程池时，需要指定TaskDecorator，在装饰代码中获取到ThreadLocal的值。
@Configuration
@EnableAsync
public class ThreadPoolConfig
@Bean("asyncServiceExecutor")
public ThreadPoolTaskExecutor asyncRabbitTimeoutServiceExecutor()
ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
//核心线程数
threadPoolTaskExecutor.setCorePoolSize(5);
//核心线程若处于闲置状态的话，超过一定的时间(KeepAliveTime)，就会销毁掉。
threadPoolTaskExecutor.setAllowCoreThreadTimeOut(true);
//最大线程数
threadPoolTaskExecutor.setMaxPoolSize(10);
//配置队列大小
threadPoolTaskExecutor.setQueueCapacity(300);
//加入装饰器
threadPoolTaskExecutor.setTaskDecorator(new ContextCopyingDecorator());
//配置线程池前缀
threadPoolTaskExecutor.setThreadNamePrefix("test-log-");
//拒绝策略:只要线程池未关闭，该策略直接在调用者线程中串行运行被丢弃的任务，显然这样不会真的丢弃任务，但是可能会造成调用者性能急剧下降
threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
threadPoolTaskExecutor.initialize();
return threadPoolTaskExecutor;
@Slf4j
static class ContextCopyingDecorator implements TaskDecorator
@Nonnull
@Override
public Runnable decorate(@Nonnull Runnable runnable)
//主流程
String res = TestMappingController.local.get();
log.info("装饰前：" + res);
//子线程逻辑
return () ->
try
//将变量重新放入到run线程中。
TestMappingController.local.set(res);
log.info("打印日志-开始");
runnable.run();
finally
log.info("打印日志-结束");
;
业务逻辑：
@PostMapping("test2")
public void test()
//主线程执行
local.set("主线程设置的ThreadLocal");
//子线程执行
executor.execute(() - >
String res = local.get();
log.info("线程:" + res);
);
执行结果：可以看到，在ContextCopyingDecorator中是主线程在调用，所以可以获取到主线程的ThreadLocal信息。
2020-11-17 18:54:25,027 INFO [1747069] [http-nio-8087-exec-5] [] (ThreadPoolConfig.java:53): 装饰前：主线程设置的ThreadLocal
2020-11-17 18:54:25,028 INFO [1747070] [test-log-4] [] (ThreadPoolConfig.java:59): 打印日志-开始
2020-11-17 18:54:25,028 INFO [1747070] [test-log-4] [] (TestMappingController.java:46): 线程:主线程设置的ThreadLocal
2020-11-17 18:54:25,028 INFO [1747070] [test-log-4] [] (ThreadPoolConfig.java:62): 打印日志-结束
以上是关于多线程调用如何传递请求上下文？简述ThreadLocal和TaskDecorator的主要内容，如果未能解决你的问题，请参考以下文章