SpringMVC处理异步请求-蒲公英云

在上一篇博客中我们讲了一些线程池及异步请求的好处，这一篇我们主要讲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);
    }
}