如何在异步任务执行程序中启用请求范围

Question

在我的应用程序中,我有一些异步Web服务.服务器接受请求,返回OK响应并使用AsyncTaskExecutor启动处理请求.我的问题是如何在此处启用请求范围,因为在此处理中我需要获取注释的类:

@Scope(value = WebApplicationContext.SCOPE_REQUEST, proxyMode = ScopedProxyMode.TARGET_CLASS)

现在我得到例外:

org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'scopedTarget.requestContextImpl': 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; nested exception is 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/DispatcherPortlet: In this case, use RequestContextListener or RequestContextFilter to expose the current request.

因为它运行SimpleAsyncTaskExecutor而不是运行DispatcherServlet

我的异步处理请求

taskExecutor.execute(new Runnable() {

    @Override
    public void run() {
        asyncRequest(request);
    }
});

taskExecutor的位置是:

<bean id="taskExecutor" class="org.springframework.core.task.SimpleAsyncTaskExecutor" />

Answer 1

我们遇到了同样的问题 - 需要使用@Async在后台执行代码,因此它无法使用任何Session或RequestScope bean.我们通过以下方式解决了它:

• 创建一个自定义TaskPoolExecutor,用于存储带有任务的范围信息
• 创建一个特殊的Callable(或Runnable),它使用该信息来设置和清除后台线程的上下文
• 创建覆盖配置以使用自定义执行程序

注意:这仅适用于Session和Request范围的bean,而不适用于安全上下文(如Spring Security).如果您正在使用安全上下文,则必须使用其他方法来设置安全上下文.

注2:为简洁起见,仅显示Callable和submit()实现.您可以对Runnable和execute()执行相同的操作.

这是代码:

执行人:

public class ContextAwarePoolExecutor extends ThreadPoolTaskExecutor {
    @Override
    public <T> Future<T> submit(Callable<T> task) {
        return super.submit(new ContextAwareCallable(task, RequestContextHolder.currentRequestAttributes()));
    }

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

赎回:

public class ContextAwareCallable<T> implements Callable<T> {
    private Callable<T> task;
    private RequestAttributes context;

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

    @Override
    public T call() throws Exception {
        if (context != null) {
            RequestContextHolder.setRequestAttributes(context);
        }

        try {
            return task.call();
        } finally {
            RequestContextHolder.resetRequestAttributes();
        }
    }
}

组态:

@Configuration
public class ExecutorConfig extends AsyncConfigurerSupport {
    @Override
    @Bean
    public Executor getAsyncExecutor() {
        return new ContextAwarePoolExecutor();
    }
}

Answer 2

最简单的方法是使用这样的任务装饰器:

static class ContextCopyingDecorator implements TaskDecorator {
    @Nonnull
    @Override
    public Runnable decorate(@Nonnull Runnable runnable) {
        RequestAttributes context =
                RequestContextHolder.currentRequestAttributes();
        Map<String, String> contextMap = MDC.getCopyOfContextMap();
        return () -> {
            try {
                RequestContextHolder.setRequestAttributes(context);
                MDC.setContextMap(contextMap);
                runnable.run();
            } finally {
                MDC.clear();
                RequestContextHolder.resetRequestAttributes();
            }
        };
    }
}

要将此装饰器添加到任务执行程序,您只需将其添加到配置例程中:

@Override
@Bean
public Executor getAsyncExecutor() {
    ThreadPoolTaskExecutor poolExecutor = new ThreadPoolTaskExecutor();
    poolExecutor.setTaskDecorator(new ContextCopyingDecorator());
    poolExecutor.initialize();
    return poolExecutor;
}

不需要额外的持有者或自定义线程池任务执行程序.

Answer 3

由于原始父请求处理线程可能已将响应提交到客户端并且所有请求对象都已销毁,因此无法在子异步线程中获取请求范围对象.处理此类方案的一种方法是使用自定义范围,如SimpleThreadScope.

SimpleThreadScope的一个问题是子线程不会继承父范围变量,因为它在内部使用简单的ThreadLocal.要克服该实现自定义范围,它与SimpleThreadScope完全相似,但在内部使用InheritableThreadLocal.有关更多信息,请参阅 Spring MVC:如何在生成的线程中使用请求范围的bean？

Answer 4

之前提到的解决方案对我不起作用。解决方案不起作用的原因是，正如@Thilak 的帖子中所述，一旦原始父线程提交对客户端的响应，请求对象可能会被垃圾收集。但是通过对@Armadillo 提供的解决方案进行一些调整，我能够让它工作。我正在使用弹簧靴 2.2

这是我遵循的。

• 创建一个自定义 TaskPoolExecutor，用于存储（克隆后）任务范围内的信息。
• 创建一个特殊的 Callable（或 Runnable），它使用克隆的信息来设置当前上下文值并清除异步线程的上下文。

执行者（与@Armadillo 的帖子相同）：

public class ContextAwarePoolExecutor extends ThreadPoolTaskExecutor {
    @Override
    public <T> Future<T> submit(Callable<T> task) {
        return super.submit(new ContextAwareCallable(task, RequestContextHolder.currentRequestAttributes()));
    }

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

可调用：

public class ContextAwareCallable<T> implements Callable<T> {
  private Callable<T> task;
  private final RequestAttributes requestAttributes;

  public ContextAwareCallable(Callable<T> task, RequestAttributes requestAttributes) {
    this.task = task;
    this.requestAttributes = cloneRequestAttributes(requestAttributes);
  }

  @Override
  public T call() throws Exception {
    try {
      RequestContextHolder.setRequestAttributes(requestAttributes);
      return task.call();
    } finally {
        RequestContextHolder.resetRequestAttributes();
    }
  }

  private RequestAttributes cloneRequestAttributes(RequestAttributes requestAttributes){
    RequestAttributes clonedRequestAttribute = null;
    try{
      clonedRequestAttribute = new ServletRequestAttributes(((ServletRequestAttributes) requestAttributes).getRequest(), ((ServletRequestAttributes) requestAttributes).getResponse());
      if(requestAttributes.getAttributeNames(RequestAttributes.SCOPE_REQUEST).length>0){
        for(String name: requestAttributes.getAttributeNames(RequestAttributes.SCOPE_REQUEST)){
          clonedRequestAttribute.setAttribute(name,requestAttributes.getAttribute(name,RequestAttributes.SCOPE_REQUEST),RequestAttributes.SCOPE_REQUEST);
        }
      }
      if(requestAttributes.getAttributeNames(RequestAttributes.SCOPE_SESSION).length>0){
        for(String name: requestAttributes.getAttributeNames(RequestAttributes.SCOPE_SESSION)){
          clonedRequestAttribute.setAttribute(name,requestAttributes.getAttribute(name,RequestAttributes.SCOPE_SESSION),RequestAttributes.SCOPE_SESSION);
        }
      }
      if(requestAttributes.getAttributeNames(RequestAttributes.SCOPE_GLOBAL_SESSION).length>0){
        for(String name: requestAttributes.getAttributeNames(RequestAttributes.SCOPE_GLOBAL_SESSION)){
          clonedRequestAttribute.setAttribute(name,requestAttributes.getAttribute(name,RequestAttributes.SCOPE_GLOBAL_SESSION),RequestAttributes.SCOPE_GLOBAL_SESSION);
        }
      }
      return clonedRequestAttribute;
    }catch(Exception e){
      return requestAttributes;
    }
  }
}

我所做的更改是引入 cloneRequestAttributes() 来复制和设置 RequestAttribute，以便即使在原始父线程向客户端提交响应之后，这些值仍然可用。

配置：由于还有其他异步配置，并且我不希望该行为适用于其他异步执行程序，因此我创建了自己的任务执行程序配置。

@Configuration
@EnableAsync
public class TaskExecutorConfig {

    @Bean(name = "contextAwareTaskExecutor")
    public TaskExecutor getContextAwareTaskExecutor() {
        ContextAwarePoolExecutor taskExecutor = new ConAwarePoolExecutor();
        taskExecutor.setMaxPoolSize(20);
        taskExecutor.setCorePoolSize(5);
        taskExecutor.setQueueCapacity(100);
        taskExecutor.setThreadNamePrefix("ContextAwareExecutor-");
        return taskExecutor;
    }
}

最后在异步方法上，我使用了执行程序名称。

    @Async("contextAwareTaskExecutor")
    public void asyncMethod() {

    }

替代解决方案：

We ended up in this trouble by trying to reuse an existing component class. Though the solution made it look like it is convenient. Its much less hassle (cloning objects and reserving thread pool) if we could have referred the relevant request scoped values as method parameters. In our case, we are planning to refactor the code in such a way that the component class which is using the request scoped bean, and being reused from the async method, to accept the values as method parameters. Request scoped bean is removed from the reusable component and moved to the component class which invokes its method. To put what I just described it in code:

Our current state is :

@Async("contextAwareTaskExecutor")
    public void asyncMethod() {
       reUsableCompoment.executeLogic() //This component uses the request scoped bean.
    }

Refactored code:

    @Async("taskExecutor")
    public void asyncMethod(Object requestObject) {
       reUsableCompoment.executeLogic(requestObject); //Request scoped bean is removed from the component and moved to the component class which invokes it menthod.
    }

Answer 5

上述解决方案都不适合我，因为在我的情况下，父线程响应返回给客户端的请求，并且请求范围的对象无法在任何工作线程中引用。

我只是做了一个变通办法来使上述事情发挥作用。我正在使用 Spring Boot 2.2 并将 customTaskExecutor 与上面指定的 ContextAwareCallable 一起使用。

异步配置：

@Bean(name = "cachedThreadPoolExecutor")
public Executor cachedThreadPoolExecutor() {

    ThreadPoolTaskExecutor threadPoolTaskExecutor = new ContextAwarePoolExecutor();
    threadPoolTaskExecutor.setCorePoolSize(corePoolSize);
    threadPoolTaskExecutor.setMaxPoolSize(maxPoolSize);
    threadPoolTaskExecutor.setQueueCapacity(queueCapacity);
    threadPoolTaskExecutor.setAllowCoreThreadTimeOut(true);
    threadPoolTaskExecutor.setThreadNamePrefix("ThreadName-");
    threadPoolTaskExecutor.initialize();
    return threadPoolTaskExecutor;

}

ContextAwarePoolExecutor：

public class ContextAwarePoolExecutor extends ThreadPoolTaskExecutor {

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

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

   }

}

创建自定义上下文感知可调用：

 public class ContextAwareCallable<T> implements Callable<T> {
   private Callable<T> task;
   private CustomRequestScopeAttributes customRequestScopeAttributes;
   private static final String requestScopedBean = 
  "scopedTarget.requestScopeBeanName";

   public ContextAwareCallable(Callable<T> task, RequestAttributes context) {
    this.task = task;
    if (context != null) {
       //This is Custom class implements RequestAttributes class
        this.customRequestScopeAttributes = new CustomRequestScopeAttributes();

        //Add the request scoped bean to Custom class       
        customRequestScopeAttributes.setAttribute
        (requestScopedBean,context.getAttribute(requestScopedBean,0),0);
        //Set that in RequestContextHolder and set as Inheritable as true 
       //Inheritable is used for setting the attributes in diffrent ThreadLocal objects.
        RequestContextHolder.setRequestAttributes
           (customRequestScopeAttributes,true);
     }
 }

   @Override
   public T call() throws Exception {
   try {
      return task.call();
    } finally {
        customRequestScopeAttributes.removeAttribute(requestScopedBean,0);
    }
   }
}

定制类：

public class CustomRequestScopeAttributes implements RequestAttributes { 

  private Map<String, Object> requestAttributeMap = new HashMap<>();
  @Override
  public Object getAttribute(String name, int scope) {
    if(scope== RequestAttributes.SCOPE_REQUEST) {
        return this.requestAttributeMap.get(name);
    }
    return null;
}
@Override
public void setAttribute(String name, Object value, int scope) {
    if(scope== RequestAttributes.SCOPE_REQUEST){
        this.requestAttributeMap.put(name, value);
    }
}
@Override
public void removeAttribute(String name, int scope) {
    if(scope== RequestAttributes.SCOPE_REQUEST) {
        this.requestAttributeMap.remove(name);
    }
}
@Override
public String[] getAttributeNames(int scope) {
    if(scope== RequestAttributes.SCOPE_REQUEST) {
        return this.requestAttributeMap.keySet().toArray(new String[0]);
    }
    return  new String[0];
 }
 //Override all methods in the RequestAttributes Interface.

}

最后在需要的方法中添加Async注解。

  @Async("cachedThreadPoolExecutor")    
  public void asyncMethod() {     
     anyService.execute() //This Service execution uses request scoped bean
  }