Spring异步请求处理

系统管理员 2023-07-24 05:34 18阅读 0赞

### Servlet容器配置 ###

**在web.xml中对DispatcherServlet和所有filter添加**

对于配置了web.xml的应用程序，请确保更新至版本3.0：

<web-app xmlns="http://java.sun.com/xml/ns/javaee"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    http://java.sun.com/xml/ns/javaee
    http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
    version="3.0">
    
    </web-app

必须通过web.xml中的 true </ async-supported>子元素在DispatcherServlet上启用异步支持。 另外，必须将参与异步请求处理的所有Filter配置为支持ASYNC调度程序类型。 为Spring框架提供的所有过滤器启用ASYNC调度程序类型应该是安全的，因为它们通常扩展了OncePerRequestFilter，并且可以在运行时检查是否需要将过滤器包含在异步调度中。

以下是一些示例web.xml配置：

<web-app xmlns="http://java.sun.com/xml/ns/javaee" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation=" http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd" version="3.0">
        
    <filter>
       <filter-name>Spring OpenEntityManagerInViewFilter</filter-name>
       <filter-class>org.springframework.orm.jpa.support.OpenEntityManagerInViewFilter</filter-class>
       <async-supported>true</async-supported>
    </filter>
    <filter-mapping>
         <filter-name>Spring OpenEntityManagerInViewFilter</filter-name>
         <url-pattern>/*</url-pattern>
         <dispatcher>REQUEST</dispatcher>
         <dispatcher>ASYNC</dispatcher>
    </filter-mapping>
    <servlet>
         <servlet-name>dispatcher</servlet-name>
         <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
            <init-param>
                <param-name>contextConfigLocation</param-name>
                <param-value/>
            </init-param>
            <load-on-startup>2</load-on-startup>
            <async-supported>true</async-supported>
        </servlet>
         <servlet-mapping>
            <servlet-name>dispatcher</servlet-name>
            <url-pattern>/*</url-pattern>
         </servlet-mapping>
    </web-app>

注意：如果你的Filter是基于注解配置的需要增加如下，**@WebFilter(asyncSupported = true,dispatcherTypes = \{DispatcherType.ASYNC,DispatcherType.REQUEST\})**

如果使用Servlet 3（例如通过WebApplicationInitializer的基于Java的配置），则还需要设置“ asyncSupported”标志以及ASYNC调度程序类型，就像使用web.xml一样。为了简化所有配置，请考虑扩展AbstractDispatcherServletInitializer，或更好的AbstractAnnotationConfigDispatcherServletInitializer，它会自动设置这些选项并使注册Filter实例非常容易。

### Spring MVC 配置 ###

MVC Java配置和MVC名称空间提供用于配置异步请求处理的选项。 WebMvcConfigurer具有方法configureAsyncSupport，而<mvc：annotation-driven>有一个子元素

这些允许配置用于异步请求的默认超时值，如果未设置，则取决于底层的Servlet容器（例如，在Tomcat上为10秒）。 您还可以配置AsyncTaskExecutor来执行从控制器方法返回的Callable实例。 强烈建议配置此属性，因为默认情况下，Spring MVC使用SimpleAsyncTaskExecutor。 它不会重复使用线程，因此不建议用于生产环境。MVC Java配置和MVC命名空间还允许您注册CallableProcessingInterceptor和DeferredResultProcessingInterceptor实例。

> default-timeout:指定异步请求处理超时之前的时间（以毫秒为单位）。在Servlet 3中，超时从主要请求处理线程退出后开始，到请求结束时结束再次分派以进一步处理同时产生的结果。 如果未设置此值，使用底层实现的默认超时时间，例如 使用Servlet 3在Tomcat上运行10秒。

### Java Config配置 ###

/** * 异步配置类 */
    @Configuration
    public class AsynWebConfig implements WebMvcConfigurer { 
    
        //配置自定义TaskExecutor
        @Override
        public void configureAsyncSupport(AsyncSupportConfigurer configurer) { 
            configurer.setDefaultTimeout(60 * 1000L);
            configurer.registerCallableInterceptors(timeoutInterceptor());
            configurer.setTaskExecutor(threadPoolTaskExecutor());
        }
    
        //异步处理拦截
        @Bean
        public TimeoutCallableProcessingInterceptor timeoutInterceptor() { 
            return new TimeoutCallableProcessingInterceptor();
        }
        //异步线程池
        @Bean
        public ThreadPoolTaskExecutor threadPoolTaskExecutor() { 
            ThreadPoolTaskExecutor t = new ThreadPoolTaskExecutor();
            t.setCorePoolSize(5);
            t.setMaxPoolSize(10);
            t.setThreadNamePrefix("NEAL");
            return t;
        }
    
    }

### 配置异步请求处理 ###

Spring MVC 3.2引入了基于Servlet 3的异步请求处理。 现在，控制器方法无需像往常一样返回值，而是可以返回java.util.concurrent.Callable并从Spring MVC托管线程产生返回值。 同时，退出并释放主要的Servlet容器线程，并允许其处理其他请求。 Spring MVC借助TaskExecutor在一个单独的线程中调用Callable，当Callable返回时，该请求被分派回Servlet容器，以使用Callable返回的值恢复处理。 这是这种控制器方法的示例：

@PostMapping(value = "v1/files.do")
        public Callable<String> processUpload(final MultipartFile file) { 
            return new Callable<String>() { 
                @Override
                public String call() throws Exception { 
    
                    return "someView";
                }
            };
        }

另一个选项是控制器方法返回DeferredResult的一个实例。在这种情况下，返回值也会从任何线程中产生，即一个不是由Spring MVC管理的线程。例如，可能会在响应某些外部事件(如JMS消息、调度任务等)时生成结果。下面是这样一个控制器方法的例子:

使用阻塞队列异步处理用户请求，超过阻塞队列容量提示限流

import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.*;
    import org.springframework.web.context.request.async.DeferredResult;
    
    import java.util.UUID;
    
    /** * @author Created by niugang on 2020/4/2/20:00 */
    @RestController
    @Slf4j
    public class DeferredResultUserInfoSaveController { 
    
        private final SimilarQueueHolder similarQueueHolder;
    
        @Autowired
        public DeferredResultUserInfoSaveController(SimilarQueueHolder similarQueueHolder) { 
            this.similarQueueHolder = similarQueueHolder;
        }
    
        @PostMapping("/deferred/result")
        public DeferredResult<Object> deferredResultHelloWolrd(@RequestBody UserInfo userInfo ) { 
    
            printlnThread("主线程--deferredResultHelloWolrd开始执行");
            //声明异步DeferredResult
            DeferredResult<Object> deferredResult = new DeferredResult<>();
            userInfo.setId(UUID.randomUUID().toString());
            deferredResult.setResult(userInfo);
            //模拟放入消息队列
            boolean offer = similarQueueHolder.getBlockingDeque().offer(deferredResult);
            if(!offer){ 
                log.info("添加任务到队列：{}",offer);
                DeferredResult<Object> deferredResult1 = new DeferredResult<>();
                deferredResult1.setResult("限流了稍后重试");
                return deferredResult1;
            }
            log.info("添加任务到队列：{}",offer);
            printlnThread("主线程--deferredResultHelloWolrd结束执行");
            return deferredResult;
        }
    
    
    
        /** * 打印当前线程 * @param object object */
        private void printlnThread(Object object) { 
            String threadName = Thread.currentThread().getName();
            log.info("HelloWorldAsyncController[{}]:{} ",threadName,object) ;
        }
    
    
    
    }

import lombok.Data;
    
    /** * @author Created by niugang on 2020/4/2/20:04 */
    @Data
    public class UserInfo { 
    
        private  String name;
    
    
        private int  age;
    
        private String id;
    }

import org.springframework.stereotype.Component;
    import org.springframework.web.context.request.async.DeferredResult;
    
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    
    /** * 模拟消息队列 * * @author Created by niugang on 2020/4/2/19:59 */
    @Component
    public class SimilarQueueHolder { 
    
        /** * 创建容量为5的阻塞队列 */
        private static BlockingQueue<DeferredResult<Object>> blockingDeque = new ArrayBlockingQueue<>(5);
    
        public BlockingQueue<DeferredResult<Object>> getBlockingDeque() { 
            return blockingDeque;
        }
    
    }

import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.ApplicationListener;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.event.ContextRefreshedEvent;
    import org.springframework.web.context.request.async.DeferredResult;
    
    import java.util.concurrent.TimeUnit;
    
    /** * 使用监听器来模拟消息队列处理 * @author Created by niugang on 2020/4/2/20:00 */
    @Configuration
    @Slf4j
    public class QueueListener implements ApplicationListener<ContextRefreshedEvent> { 
    
        private final SimilarQueueHolder similarQueueHolder;
    
        @Autowired
        public QueueListener(SimilarQueueHolder similarQueueHolder) { 
            this.similarQueueHolder = similarQueueHolder;
        }
    
        @Override
        public void onApplicationEvent(ContextRefreshedEvent event) { 
            new Thread(()->{ 
                while(true) { 
                    try { 
                        //从队列中取出DeferredResult
                        DeferredResult<Object> deferredResult = similarQueueHolder.getBlockingDeque().take();
                        log.info("开始DeferredResult异步处理");
                        //模拟处理时间
                        TimeUnit.SECONDS.sleep(3);
                        log.info("用户信息：{}",deferredResult.getResult());
                        log.info("结束DeferredResult异步处理");
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    
    
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pdWdhbmcwOTIw_size_16_color_FFFFFF_t_70_pic_center]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pdWdhbmcwOTIw_size_16_color_FFFFFF_t_70_pic_center 1]

如果不了解Servlet 3.0异步请求处理特性，就很难理解这一点。多了解这方面的情况肯定会有帮助。以下是关于潜在机制的一些基本事实:

*  可以通过调用request.startAsync()将ServletRequest置于异步模式。 这样做的主要效果是Servlet以及所有过滤器都可以退出，但响应将保持打开状态，以便以后可以完成处理
 *  调用request.startAsync()返回AsyncContext，该AsyncContext可用于进一步控制异步处理。 例如，它提供了方法分派，类似于Servlet API的转发，但它允许应用程序恢复Servlet容器线程上的请求处理。
 *  ServletRequest提供对当前DispatcherType的访问，该访问可用于区分处理初始请求，异步分派，转发和其他分派器类型。

考虑到上述内容，以下是使用Callable进行异步请求处理的事件序列：

*  控制器返回Callable。
 *  Spring MVC开始异步处理，并将Callable提交给TaskExecutor在单独的线程中进行处理。
 *  DispatcherServlet和所有Filter退出Servlet容器线程，但响应保持打开状态
 *  Callable产生结果，Spring MVC将请求分派回Servlet容器以恢复处理。
 *  再次调用DispatcherServlet，并使用Callable异步生成的结果恢复处理。

DeferredResult请求的事件序列

*  控制器返回DeferredResult并将其保存在一些内存队列或列表中，可以在其中访问
 *  Spring MVC开始异步处理
 *  DispatcherServlet和所有已配置的Filter退出请求处理线程，但响应保持打开状态。
 *  应用程序从某个线程设置DeferredResult，Spring MVC将请求分派回Servlet容器。
 *  再次调用DispatcherServlet，并以异步产生的结果恢复处理。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pdWdhbmcwOTIw_size_16_color_FFFFFF_t_70]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pdWdhbmcwOTIw_size_16_color_FFFFFF_t_70_pic_center]: /images/20230528/2917851fe0874ee195611994b95bd752.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pdWdhbmcwOTIw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20230528/aa2b877472d3429f8d730d884d83de97.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25pdWdhbmcwOTIw_size_16_color_FFFFFF_t_70]: /images/20230528/a94688a255b342239d47faec8a0bfe98.png