SpringMVC处理异步请求

た入场券 2022-06-06 10:43 367阅读 0赞

在上一篇博客中我们讲了一些线程池及异步请求的好处，这一篇我们主要讲SpringMVC如何使用异步请求。

在SpringMVC使用异步响应，主要是指定Controller的@RequestMapping标识的方法的返回值类型，常见有三种类型：Callable，DeferredResult，WebAsyncTask。当SpringMVC检测到返回的对象类型是这三种类型是，会启动异步形式处理。

对于异步请求，也就是需要额外的线程来代理Servlet容器内线程来处理用户请求，也就是说需要额外的线程，可以使用项目内线程池，比如Spring，可以定义一个线程池，专门用户那些项目内的异步处理及异步请求的处理。

对于异步请求，一般是用于处理时间比较长的请求，那么久需要一个时长限制，可以定义一个默认的超时时限，当然每个请求还可以定义自己的超时时限，如果请求未定义，则使用默认时限。

对于异步请求，SpringMVC有一些针对其的Inteceptor，用户可以自定义自己的异步请求拦截器，比如在超时了做些什么，在异步处理结束了做些什么，记录异步请求的处理时间等等。

下面看异步请求的环境配置，以Java形式配置来举例：

@Configuration
    @EnableWebMvc
    public class MvcContextConfig extends WebMvcConfigurerAdapter { 
    
        /** * Spring内部自定义线程池,可以对@Async注解以及Controler返回的Callable,WebAsyncTask和DeferredResult等Spring内异步线程的支持 * * 当一个任务通过execute(Runnable)方法欲添加到线程池时： * * 如果此时线程池中的数量小于corePoolSize，即使线程池中的线程都处于空闲状态，也要创建新的线程来处理被添加的任务。 * * 如果此时线程池中的数量等于 corePoolSize，但是缓冲队列 workQueue未满，那么任务被放入缓冲队列。 * * 如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量小于maximumPoolSize，建新的线程来处理被添加的任务。 * * 如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量等于maximumPoolSize，那么通过 handler所指定的策略来处理此任务。 * * 也就是：处理任务的优先级为：核心线程corePoolSize、任务队列workQueue、最大线程 maximumPoolSize，如果三者都满了，使用handler处理被拒绝的任务。 * * 当线程池中的线程数量大于 corePoolSize时，如果某线程空闲时间超过keepAliveTime，线程将被终止。这样，线程池可以动态的调整池中的线程数。 * * {@linkplain ThreadPoolExecutor#execute(Runnable)} * * @return */
        @Bean
        public ThreadPoolTaskExecutor threadPoolTaskExecutor() {
            ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
            executor.setCorePoolSize(10); // 线程池维护线程的最少数量
            executor.setMaxPoolSize(20); // 线程池维护线程的最大数量
            executor.setKeepAliveSeconds(300); // 空闲线程的最长保留时间,超过此时间空闲线程会被回收
            executor.setQueueCapacity(30); // 线程池所使用的缓冲队列
            executor.setThreadNamePrefix("Spring-ThreadPool#");
            // rejection-policy：当线程池线程已达到最大值且任务队列也满了的情况下，如何处理新任务
            // CALLER_RUNS：这个策略重试添加当前的任务，他会自动重复调用 execute() 方法，直到成功
            executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
            executor.afterPropertiesSet();
            return executor;
        }
    
        @Override
        public void configureAsyncSupport(AsyncSupportConfigurer configurer) {
            configurer.setDefaultTimeout(5 * 1000); //设置默认的超时时间
            configurer.setTaskExecutor(threadPoolTaskExecutor);  //设置异步请求使用的线程池
            //注册异步请求的拦截器
            configurer.registerCallableInterceptors(new AsyncCallableInterceptor());
            configurer.registerDeferredResultInterceptors(new AsyncDeferredResultInterceptor());
        }
    
    }

以上是基于Java形式的配置，如果是要基于配置文件的形式来配置，那么看如下代码:

public class WebMvcConfigurationSupport implements ApplicationContextAware, ServletContextAware { 
        @Bean
        public RequestMappingHandlerAdapter requestMappingHandlerAdapter() {
            ......
            AsyncSupportConfigurer configurer = new AsyncSupportConfigurer();
            configureAsyncSupport(configurer);
            if (configurer.getTaskExecutor() != null) {
                adapter.setTaskExecutor(configurer.getTaskExecutor());
            }
            if (configurer.getTimeout() != null) {
                adapter.setAsyncRequestTimeout(configurer.getTimeout());
            }
            adapter.setCallableInterceptors(configurer.getCallableInterceptors());
            adapter.setDeferredResultInterceptors(configurer.getDeferredResultInterceptors());
    
            return adapter;
        }
    }

实际的那些异步环境配置，最终都配置到了RequestMappingHandlerAdapter 这个Bean中，因此使用配置文件配置是，直接配置RequestMappingHandlerAdapter 这个Bean。

对于异步拦截器来说，Callable和WebAsyncTask共用一种拦截器CallableProcessingInterceptor，因为WebAsyncTask就是对Callable的一个封装，添加了一些额外的方法，比如超时处理等，可以通过继承CallableProcessingInterceptorAdapter适配器重写想要添加逻辑的方法。DeferredResult使用DeferredResultProcessingInterceptor作为拦截器，可以通过继承DeferredResultProcessingInterceptorAdapter适配器来重写想要添加逻辑的 方法。SpringMVC会在处理的相应阶段调用拦截器的相应方法。

下面讲这三种异步处理的使用形式。

对于Callable来说，很简单，直接返回一个Callable对象，将要处理的逻辑放在Callable的run方法内，当SpringMVC检测到返回的是Callable类型的对象时，会使用我们定义的线程池内的线程去执行此Callable，检测逻辑稍后讲。

对于DeferredResult来说，关键点是要在一个线程内调用DeferredResult的setResult（T）方法，也就是说我们需要在Controller方法内就启动线程，而不能返回一个任务由SpringMVC来启动。

WebAsyncTask和Callable类似，是对Callable的一个封装，可以增加处理超时的逻辑以及处理完成后的逻辑等。

下面看使用案例:

import java.util.concurrent.Callable;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestMethod;
    import org.springframework.web.bind.annotation.RestController;
    import org.springframework.web.context.request.async.DeferredResult;
    import org.springframework.web.context.request.async.WebAsyncTask;
    
    /** * @since 2017年6月9日 上午10:23:43 * */
    @RestController
    @RequestMapping("/async")
    public class AsyncController { 
    
        private static final String TIME_OUT_RESULT = "TIME_OUT";
        private static final String ASYNC_RESULT = "SUCCESS";
    
        final static Logger LOGGER = LoggerFactory.getLogger(AsyncController.class);
    
        @RequestMapping(value = "/callable", method = RequestMethod.GET)
        public Callable<String> getCallableMessage() {
            LOGGER.debug("请求{}方法,线程id:{}", "getAsyncMessage", Thread.currentThread().getId());
            LOGGER.debug("释放线程id:{}", Thread.currentThread().getId());
            return () -> {
                Thread.sleep(10000);
                return "异步信息";
            };
        }
    
        /** * DeferredResult延迟结果使用案例 * * @return * @throws InterruptedException */
        @RequestMapping(value = "/deferred", method = RequestMethod.GET)
        public DeferredResult<String> getDeferredResult() throws InterruptedException {
            LOGGER.debug("请求{}方法,线程id:{}", "getDeferredResult", Thread.currentThread().getId());
            final DeferredResult<String> def = new DeferredResult<String>(6000L, TIME_OUT_RESULT);
    
            def.onTimeout(() -> {
                LOGGER.error("请求处理超时");
                def.setErrorResult(TIME_OUT_RESULT);
            });
    
            def.onCompletion(() -> LOGGER.debug("请求结束"));
    
            new Thread(() -> {
                def.setResult(processAsyncResult());
                def.setResultHandler((result) -> LOGGER.debug("DeferredResultHandler.handleResult[{}]", def.getResult()));
            }).start();
    
            LOGGER.debug("释放线程id:{}", Thread.currentThread().getId());
            return def;
        }
    
        private String processAsyncResult() {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return ASYNC_RESULT;
        }
    
        /** * WebAsyncTask使用实例 * * @return */
        @RequestMapping(value = "/asyncTask", method = RequestMethod.GET)
        public WebAsyncTask<String> getAsyncTask() {
            WebAsyncTask<String> asyncTask = new WebAsyncTask<>(6000L, () -> processAsyncResult());
            asyncTask.onCompletion(() -> LOGGER.debug("异步任务结束"));
            asyncTask.onTimeout(() -> {
                LOGGER.debug("异步任务结束");
                return TIME_OUT_RESULT;
            });
            return asyncTask;
        }
    
    }

SpringMVC监测Controller方法返回的对象类型，如果是Callable,使用CallableMethodReturnValueHandler去处理；如果是DeferredResult，使用DeferredResultMethodReturnValueHandler处理；如果是WebAsyncTask，则使用AsyncTaskMethodReturnValueHandler处理。他们最终都使用WebAsyncManager这个类来处理异步请求，这个类封装了HttpServletRequest，相关内容感兴趣的可以自行查看。

下面看测试代码，测试框架可以使用我的一篇博客中将的Mockmvc，配置代码也在博客上。

public class AsyncControllerTest extends BaseControllerTest { 
    
        @Autowired
        private AsyncController asyncController;
    
        /* (non-Javadoc) * @see com.bob.test.config.BaseControllerTest#init() */
        @Override
        protected void init() {
            super.mappedController = asyncController;
        }
    
        @Test
        public void testDeferredResult() throws Exception {
            String result = this.getAsyncRequest("/async/deferred", "");
            System.out.println(result);
        }
    
    }