java - Controller 中的并行流中没有线程绑定(bind)请求错误

Question

我的 Controller 上有一个端点，它获取 ID 列表。对于每个 ID，我从 REST 端点获取详细信息，我希望异步执行此操作。这是我的实现:

@PostMapping("/leaderboard")
@ApiOperation("Get clients leaderboard statistics")
public List<Dto> clientLeaderboard(@RequestBody List<String> accountIds) {

    IntStream.range(0, accountIds.size()).parallel().forEach(i -> {
        AccountDTO accountDTO = accountMaintenanceRestClient.getAccount(accountIds.get(i));
        // More logic
    }

}

这给了我错误:

Unhandled exception occurred

java.lang.IllegalStateException: 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.

我之前实际上遇到过这个错误，并设法在答案 here 中修复它。 。同样的解决方案是否适用于此？如果是这样，我如何向这个并行流提供现有的执行器？

编辑

AccountMaintenanceRestClient 代码。我正在使用OpenFeign作为其中的一部分。

public interface AccountMaintenanceRestClient {
   @RequestLine("GET /account/{accountId}")
   @Profiled(tag = "AccountMaintenanceRestClient.{$0}.getAccountBalance", logFailuresSeparately = true)
   AccountDTO getAccount(@Param("accountId") String accountId);

   @RequestLine("GET /referencedata/statuscode")
   @Profiled(tag = "AccountMaintenanceRestClient.getStatusCodes", logFailuresSeparately = true)
   Map<String, String> getStatusCodes();

   @RequestLine("GET /referencedata/websites")
   @Profiled(tag = "AccountMaintenanceRestClient.getWebsitesIds", logFailuresSeparately = true)
   List<WebSiteDTO> getWebsitesIds();
}

堆栈跟踪:

java.lang.IllegalStateException: 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.
    at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
    at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
    at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
    at java.util.concurrent.ForkJoinTask.getThrowableException(ForkJoinTask.java:593)
    at java.util.concurrent.ForkJoinTask.reportException(ForkJoinTask.java:677)
    at java.util.concurrent.ForkJoinTask.invoke(ForkJoinTask.java:735)
    at java.util.stream.ForEachOps$ForEachOp.evaluateParallel(ForEachOps.java:160)
    at java.util.stream.ForEachOps$ForEachOp$OfInt.evaluateParallel(ForEachOps.java:189)
    at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:233)
    at java.util.stream.IntPipeline.forEach(IntPipeline.java:404)
    at java.util.stream.IntPipeline$Head.forEach(IntPipeline.java:560)
    at com.foobar.wt.igip.gateway.web.controllers.surge.paris.SurgeParisController.clientLeaderboard(SurgeParisController.java:58)
    at com.foobar.wt.igip.gateway.web.controllers.surge.paris.SurgeParisController$$FastClassBySpringCGLIB$$a143909.invoke(<generated>)
    at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:204)
    at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:667)
    at com.foobar.wt.igip.gateway.web.controllers.surge.paris.SurgeParisController$$EnhancerBySpringCGLIB$$67cb3a1e.clientLeaderboard(<generated>)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:205)
    at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:133)
    at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:97)
    at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:854)
    at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:765)
    at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:85)
    at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:967)
    at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:901)
    at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:970)
    at org.springframework.web.servlet.FrameworkServlet.doPost(FrameworkServlet.java:872)
    at javax.servlet.http.HttpServlet.service(HttpServlet.java:650)
    at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:846)
    at javax.servlet.http.HttpServlet.service(HttpServlet.java:731)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:303)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)
    at org.apache.tomcat.websocket.server.WsFilter.doFilter(WsFilter.java:52)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)
    at org.springframework.boot.web.filter.ApplicationContextHeaderFilter.doFilterInternal(ApplicationContextHeaderFilter.java:54)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)
    at org.springframework.boot.actuate.trace.WebRequestTraceFilter.doFilterInternal(WebRequestTraceFilter.java:111)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)
    at uk.co.igindex.commons.bluegreen.IGClusterDetailsFilter.doFilter(IGClusterDetailsFilter.java:46)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)
    at org.springframework.security.web.FilterChainProxy.doFilterInternal(FilterChainProxy.java:208)
    at org.springframework.security.web.FilterChainProxy.doFilter(FilterChainProxy.java:177)
    at org.springframework.web.filter.DelegatingFilterProxy.invokeDelegate(DelegatingFilterProxy.java:347)
    at org.springframework.web.filter.DelegatingFilterProxy.doFilter(DelegatingFilterProxy.java:263)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)
    at org.springframework.web.filter.CharacterEncodingFilter.doFilterInternal(CharacterEncodingFilter.java:197)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)
    at org.springframework.boot.actuate.autoconfigure.MetricsFilter.doFilterInternal(MetricsFilter.java:100)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)
    at org.springframework.boot.web.support.ErrorPageFilter.doFilter(ErrorPageFilter.java:128)
    at org.springframework.boot.web.support.ErrorPageFilter.access$000(ErrorPageFilter.java:66)
    at org.springframework.boot.web.support.ErrorPageFilter$1.doFilterInternal(ErrorPageFilter.java:103)
    at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107)
    at org.springframework.boot.web.support.ErrorPageFilter.doFilter(ErrorPageFilter.java:121)
    at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:241)
    at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:208)
    at org.apache.catalina.core.StandardWrapperValve.invoke(StandardWrapperValve.java:220)
    at org.apache.catalina.core.StandardContextValve.invoke(StandardContextValve.java:122)
    at org.apache.catalina.authenticator.AuthenticatorBase.invoke(AuthenticatorBase.java:505)
    at org.apache.catalina.core.StandardHostValve.invoke(StandardHostValve.java:170)
    at org.apache.catalina.valves.ErrorReportValve.invoke(ErrorReportValve.java:103)
    at org.apache.catalina.valves.AccessLogValve.invoke(AccessLogValve.java:956)
    at org.apache.catalina.core.StandardEngineValve.invoke(StandardEngineValve.java:116)
    at org.apache.catalina.connector.CoyoteAdapter.service(CoyoteAdapter.java:423)
    at org.apache.coyote.http11.AbstractHttp11Processor.process(AbstractHttp11Processor.java:1079)
    at org.apache.coyote.AbstractProtocol$AbstractConnectionHandler.process(AbstractProtocol.java:625)
    at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1757)
    at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.run(NioEndpoint.java:1716)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
    at java.lang.Thread.run(Thread.java:748)
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/DispatcherPortlet: 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.support.WebApplicationContextUtils.currentRequestAttributes(WebApplicationContextUtils.java:309)
    at org.springframework.web.context.support.WebApplicationContextUtils.access$400(WebApplicationContextUtils.java:64)
    at org.springframework.web.context.support.WebApplicationContextUtils$RequestObjectFactory.getObject(WebApplicationContextUtils.java:325)
    at org.springframework.web.context.support.WebApplicationContextUtils$RequestObjectFactory.getObject(WebApplicationContextUtils.java:320)
    at org.springframework.beans.factory.support.AutowireUtils$ObjectFactoryDelegatingInvocationHandler.invoke(AutowireUtils.java:307)
    at com.sun.proxy.$Proxy134.getHeader(Unknown Source)
    at com.foobar.wt.igip.gateway.support.FeignSSOTokenRequestInterceptor.apply(FeignSSOTokenRequestInterceptor.java:22)
    at feign.SynchronousMethodHandler.targetRequest(SynchronousMethodHandler.java:158)
    at feign.SynchronousMethodHandler.executeAndDecode(SynchronousMethodHandler.java:88)
    at feign.SynchronousMethodHandler.invoke(SynchronousMethodHandler.java:76)
    at feign.ReflectiveFeign$FeignInvocationHandler.invoke(ReflectiveFeign.java:103)
    at com.sun.proxy.$Proxy156.getAccount(Unknown Source)
    at com.foobar.wt.igip.gateway.web.controllers.surge.paris.SurgeParisController.lambda$clientLeaderboard$2(SurgeParisController.java:59)
    at java.util.stream.ForEachOps$ForEachOp$OfInt.accept(ForEachOps.java:205)
    at java.util.stream.Streams$RangeIntSpliterator.forEachRemaining(Streams.java:110)
    at java.util.Spliterator$OfInt.forEachRemaining(Spliterator.java:693)
    at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
    at java.util.stream.ForEachOps$ForEachTask.compute(ForEachOps.java:291)
    at java.util.concurrent.CountedCompleter.exec(CountedCompleter.java:731)
    at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:289)
    at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1056)
    at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1692)
    at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:157)

FeignSSOTokenRequestInterceptor:

import feign.RequestInterceptor;
import feign.RequestTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.servlet.http.HttpServletRequest;

@Component
public class FeignSSOTokenRequestInterceptor implements RequestInterceptor {
   private static final String SSO_HEADER_NAME = "X-SECURITY-TOKEN";
   private final HttpServletRequest request;

   @Autowired
   public FeignSSOTokenRequestInterceptor(HttpServletRequest request) {
      this.request = request;
   }

   @Override
   public void apply(RequestTemplate requestTemplate) {
      requestTemplate.header(SSO_HEADER_NAME, request.getHeader(SSO_HEADER_NAME));
   }
}

Answer 1

您的 FeignSSOTokenRequestInterceptor 取决于 HttpServletRequest 代理 bean。

请求通过线程本地公开，因此线程池中的线程不知道/意识到它，因此当它尝试从线程本地映射读取请求时会发生异常。

所以基本上问题缩小到如何将请求设置到线程池的线程上下文。

像您引用的帖子一样设置线程池，然后提交任务，

@Autowired
ThreadPoolTaskExecutor poolExecutor;

CountdownLatch counter = new CountdownLatch(accountIds.size());
IntStream.range(0, accountIds.size()) // Note that .parallel is omitted
         .forEach(() -> poolExecutor.submit(() -> {
             try {
                your task
             } finally {
                counter.countDown();
             }
         })) 
counter.await(); // Wait for all tasks complete

这是首选方式，因为默认并行流使用公共(public)共享池，因此您不应该将具有繁重 IO 的任务(例如 API 访问)放置在该任务中)