Web服务请求的几种异步处理方式

标签：Web 请求异步几种 Callable 线程 DeferredResult public

我们先通过下面两张图来看下网络Web请求的异步处理和同步请求处理的区别：

在上面两个流程图中有三个角色：客户端、Web容器和业务后端服务。

两个流程中客户端对Web容器的请求，都是同步的。因为它们在请求客户端时都处于阻塞等待状态（涉及到用户态和内核态的切换），并没有进行异步处理。

在Web容器部分，第一个流程采用同步请求，第二个流程采用异步回调的形式。

通过异步处理，可以先释放容器分配给请求的线程与相关资源，减轻系统负担，从而增加了Web服务器对客户端请求的吞吐量。但并发请求量较大时，也通常要配合通过负载均衡的方案来解决，而不只是异步。

1、基于Servlet方式实现异步请求

下面是一个基于Servlet方式的异步请求示例：

@GetMapping(value = "/user/get")
public void servletReq(HttpServletRequest request) {
AsyncContext asyncContext = request.startAsync();
// 设置监听器:可设置其开始、完成、异常、超时等事件的回调处理
asyncContext.addListener(new AsyncListener() {
    @Override
    public void onTimeout(AsyncEvent event) {
        System.out.println("处理超时了...");
    }

    @Override
    public void onStartAsync(AsyncEvent event) {
        System.out.println("线程开始执行");
    }

    @Override
    public void one rror(AsyncEvent event) {
        System.out.println("执行过程中发生错误：" + event.getThrowable().getMessage());
    }

    @Override
    public void onComplete(AsyncEvent event) {
        System.out.println("执行完成，释放资源");
    }
});
//设置超时时间
asyncContext.setTimeout(6000);
asyncContext.start(new Runnable() {
    @Override
    public void run() {
        try {
            Thread.sleep(5000);
            System.out.println("内部线程：" + Thread.currentThread().getName());
            asyncContext.getResponse().getWriter().println("async processing");
        } catch (Exception e) {
            System.out.println("异步处理发生异常：" + e.getMessage());
        }
        // 异步请求完成通知，整个请求完成
        asyncContext.complete();
    }
});
//此时request的线程连接已经释放了
System.out.println("主线程：" + Thread.currentThread().getName());
}

可以看出，上述代码先执行完了主线程，也就是程序的最后一行代码的日志打印，然后才是内部线程的执行。内部线程执行完成，AsyncContext 的onComplete方法会被调用。

Servlet 3.0之前，Servlet采用Thread-Per-Request的方式处理请求，即每一次Http请求都由一个线程从头到尾处理。当涉及到耗时操作时，性能问题便比较明显。

Servlet 3.0中提供了异步处理请求。可以先释放容器分配给请求的线程与相关资源，减轻系统负担，从而增加服务的吞吐量。

Servlet 3.0的异步是通过AsyncContext对象来完成的，它可以从当前线程传给另一个线程，并归还初始线程。新的线程处理完业务可以直接返回结果给客户端。

不仅可以通过AsyncContext获取Request和Response等信息，还可以设置异步处理超时时间。通常，超时时间（单位毫秒）是需要设置的，不然无限等下去不就与同步处理一样了。

通过AsyncContext的addListener还可以添加监听事件，用来处理异步线程的开始、完成、异常、超时等事件回调。

addListener方法的参数AsyncListener的源码如下：

public interface AsyncListener extends EventListener {
    // 异步执行完毕时调用
    void onComplete(AsyncEvent var1) throws IOException;
    // 异步线程执行超时调用
    void onTimeout(AsyncEvent var1) throws IOException;
    // 异步线程出错时调用
    void one rror(AsyncEvent var1) throws IOException;
    // 异步线程开始时调用
    void onStartAsync(AsyncEvent var1) throws IOException;
}

2、基于Callable方式实现异步请求

如果使用SpringMVC轻量级Web开发框架，可以基于JDK包提供的Callable实现，具体处理流程如下：

Spring MVC开启副线程处理业务(将Callable提交到TaskExecutor)；
DispatcherServlet和所有的Filter退出Web容器的线程，但是response保持打开状态；
Callable返回结果，SpringMVC将原始请求重新派发给容器(再重新请求一次/uri)，恢复之前的处理；
DispatcherServlet重新被调用，将结果返回给用户；

下面是一个基于Callable 方式的异步请求示例：

@GetMapping("/uri")
public Callable<String> test() {
    System.out.println("主线程开始：" + Thread.currentThread().getName());
    Callable<String> result = () -> {
        System.out.println("副线程开始：" + Thread.currentThread().getName());
        Thread.sleep(1000);
        System.out.println("副线程返回：" + Thread.currentThread().getName());
        return "success";
    };
    System.out.println("主线程返回：" + Thread.currentThread().getName());
    return result;
}

Callable默认使用SimpleAsyncTaskExecutor类来执行，这个类非常简单而且没有重用线程，每次都会开一个线程去执行。在实践中，需要使用AsyncTaskExecutor类来对线程进行配置。Callable所在的包是：java.util.concurrent；

这里通过实现WebMvcConfigurer接口来完成线程池的配置：

@Configuration
public class WebConfig implements WebMvcConfigurer {
 
    @Resource
    private ThreadPoolTaskExecutor myThreadPoolTaskExecutor;
 
    /**
     * 配置线程池
     */
    @Bean(name = "asyncPoolTaskExecutor")
    public ThreadPoolTaskExecutor getAsyncThreadPoolTaskExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(2);
        taskExecutor.setMaxPoolSize(10);
        taskExecutor.setQueueCapacity(25);
        taskExecutor.setKeepAliveSeconds(200);
        taskExecutor.setThreadNamePrefix("thread-pool-");
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        taskExecutor.initialize();
        return taskExecutor;
    }
 
    @Override
    public void configureAsyncSupport(final AsyncSupportConfigurer configurer) {
        // 处理callable超时
        configurer.setDefaultTimeout(60 * 1000);
        configurer.setTaskExecutor(myThreadPoolTaskExecutor);
        configurer.registerCallableInterceptors(timeoutCallableProcessingInterceptor());
    }
 
    @Bean
    public TimeoutCallableProcessingInterceptor timeoutCallableProcessingInterceptor() {
        return new TimeoutCallableProcessingInterceptor();
    }
}

3、基于WebAsyncTask方式实现异步请求

Spring提供的WebAsyncTask是对Callable的包装，提供了更强大的功能，比如：业务处理超时回调、错误回调、完成回调等。

下面是一个基于WebAsyncTask 方式的异步请求示例：

@GetMapping("/webAsyncTask")
public WebAsyncTask<String> webAsyncTask() {
    log.info("外部线程：" + Thread.currentThread().getName());
    WebAsyncTask<String> result = new WebAsyncTask<>(60 * 1000L, new Callable<String>() {
        @Override
        public String call() {
            log.info("内部线程：" + Thread.currentThread().getName());
            return "success";
        }
    });
 
    result.onTimeout(new Callable<String>() {
        @Override
        public String call() {
            log.info("timeout callback");
            return "timeout callback";
        }
    });
 
    result.onCompletion(new Runnable() {
        @Override
        public void run() {
            log.info("finish callback");
        }
    });
    return result;
}

WebAsyncTask本质上是对Callable的封装，提供了一些事件回调的处理。WebAsyncTask所在的包：org.springframework.web.context.request.async;

4、基于DeferredResult方式实现异步请求

DeferredResult 使用方式与Callable类似，但在返回结果时不一样，它返回的时实际结果可能没有生成，实际的结果可能会在另外的线程里面设置到DeferredResult中去。

DeferredResult 的这个特性对实现服务端推技术、订单过期时间处理、长轮询、模拟MQ的功能等高级应用非常重要。

DeferredResult这个类也是在spring-web包中：org.springframework.web.context.request.async。DeferredResult的调用并不一定在Spring MVC当中，它可以是别的线程。官方的解释也是如此：

In this case the return value will also be produced from a separate thread. However, that thread is not known to Spring MVC. For example the result may be produced in response to some external event such as a JMS message, a scheduled task, etc.

也就是说，DeferredResult返回的结果也可能是由MQ、定时任务或其他线程触发。

下面是一个基于DeferredResult 方式的异步请求示例：

@Controller
@RequestMapping("/async/controller")
public class AsyncHelloController {
 
    private List<DeferredResult<String>> deferredResultList = new ArrayList<>();
 
    @ResponseBody
    @GetMapping("/hello")
    public DeferredResult<String> helloGet() throws Exception {
        DeferredResult<String> deferredResult = new DeferredResult<>();
 
        //先存起来，等待触发
        deferredResultList.add(deferredResult);
        return deferredResult;
    }
 
    @ResponseBody
    @GetMapping("/setHelloToAll")
    public void helloSet() throws Exception {
        // 让所有hold住的请求给与响应
        deferredResultList.forEach(d -> d.setResult("say hello to all"));
    }
}

第一个请求/hello，会先将deferredResult存起来，前端页面是一直等待（转圈）状态。直到发第二个请求：setHelloToAll，所有的相关页面才会有响应。

整个执行流程如下：

Controller返回一个DeferredResult，把它保存到内存里或者List里面（供后续访问）；
Spring MVC调用request.startAsync()，开启异步处理；与此同时将DispatcherServlet里的拦截器、Filter等等都马上退出主线程，但是response仍然保持打开的状态；
应用通过另外一个线程（可能是MQ消息、定时任务等）给DeferredResult#setResult值。然后SpringMVC会把这个请求再次派发给servlet容器；
DispatcherServlet再次被调用，然后处理后续的标准流程；

通过上述流程可以发现：利用DeferredResult可实现一些长连接的功能，比如当某个操作是异步时，可以先保存对应的DeferredResult对象，当异步通知回来时，再找到这个DeferredResult对象，在setResult处理结果即可。从而提高性能。

更多优质技术分享，请关注下面

标签：Web,请求,异步,几种,Callable,线程,DeferredResult,public
From： https://blog.csdn.net/mzhlqy/article/details/139854669

Web服务请求的几种异步处理方式

1、基于Servlet方式实现异步请求

2、基于Callable方式实现异步请求

3、基于WebAsyncTask方式实现异步请求

4、基于DeferredResult方式实现异步请求

相关文章

赞助商

阅读排行