SpringMVC异步处理操作(Callable和DeferredResult)

SpringMVC异步处理操作(Callable和DeferredResult)

官方文档中说DeferredResult和Callable都是为了异步生成返回值提供基本的支持。简单来说就是一个请求进来，如果你使用了DeferredResult或者Callable，在没有得到返回数据之前，DispatcherServlet和所有Filter就会退出Servlet容器线程，但响应保持打开状态，一旦返回数据有了，这个DispatcherServlet就会被再次调用并且处理，以异步产生的方式，向请求端返回值。

这么做的好处就是请求不会长时间占用服务连接池，提高服务器的吞吐量。

Callable

@GetMapping("/callable")

public Callable testCallable() throws InterruptedException {

log.info("主线程开始！");

Callable result = new Callable() {

@Override

public String call() throws Exception {

log.info("副线程开始！");

Thread.sleep(1000);

log.info("副线程结束！");

return "SUCCESS";

}

};

log.info("主线程结束！");

return result;

}

输出的结果如下：

主线程开始!

主线程结束!

副线程开始!

副线程结束!

主线程会提前返回，可以处理其他请求，等待有结果之后再输出结果

DeferredResult

一旦启用了异步请求处理功能 ，控制器就可以将返回值包装在DeferredResult，控制器可以从不同的线程异步产生返回值。优点就是可以实现两个完全不相干的线程间的通信。

我们模拟如下场景：

由于消息队列和应用2的部分太繁琐，我们使用一个类来代替

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.stereotype.Component;

@Component

public class MockQueue {

private String placeOrder;

private String completeOrder;

private Logger logger = LoggerFactory.getLogger(getClass());

public String getPlaceOrder() {

return placeOrder;

}

public void setPlaceOrder(String placeOrder) throws Exception {

new Thread(() -> {

logger.info("接到下单请求, " + placeOrder);

try {

Thread.sleep(1000);

} catch (Exception e) {

e.printStackTrace();

}

this.completeOrder = placeOrder;

logger.info("下单请求处理完毕," + placeOrder);

}).start();

}

public String getCompleteOrder() {

return completeOrder;

}

public void setCompleteOrder(String completeOrder) {

this.completeOrder = completeOrder;

}

}

定义一个Controller即线程1的部分

import org.apache.commons.lang.RandomStringUtils;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RestController;

import org.springframework.web.context.request.async.DeferredResult;

@RestController

public class AsyncController {

@Autowired

private MockQueue mockQueue;

@Autowired

private DeferredResultHolder deferredResultHolder;

private Logger logger = LoggerFactory.getLogger(getClass());

@RequestMapping("/order")

public DeferredResult order() throws Exception {

logger.info("主线程开始");

String orderNumber = RandomStringUtils.randomNumeric(8);

mockQueue.setPlaceOrder(orderNumber);

DeferredResult result = new DeferredResult<>();

deferredResultHolder.getMap().put(orderNumber, result);

return result;

}

}

定义一个类，用来线程1和线程2之间通信的，使用@Component默认为单例，方便通信。

import java.util.HashMap;

import java.util.Map;

import org.springframework.stereotype.Component;

import org.springframework.web.context.request.async.DeferredResult;

@Component

public class DeferredResultHolder {

private Map> map = new HashMap>();

public Map> getMap() {

return map;

}

public void setMap(Map> map) {

this.map = map;

}

}

定义一个类来监听订单是否处理完，如果处理完了的话就设置deferredResultHolder中的DeferredResult的值，就会返回结果了。

import org.apache.commons.lang.StringUtils;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.context.ApplicationListener;

import org.springframework.context.event.ContextRefreshedEvent;

import org.springframework.stereotype.Component;

@Component

public class QueueListener implements ApplicationListener {

@Autowired

private MockQueue mockQueue;

@Autowired

private DeferredResultHolder deferredResultHolder;

private Logger logger = LoggerFactory.getLogger(getClass());

@Override

public void onApplicationEvent(ContextRefreshedEvent event) {

new Thread(() -> {

while (true) {

if (StringUtils.UMzLVisNotBlank(mockQueue.getCompleteOrder())) {

String orderNumber = mockQueue.getCompleteOrder();

logger.info("返回订单处理结果:"+orderNumber);

deferredResultHolder.getMap().get(orderNumber).setResult("place order success");

mockQueue.setCompleteOrder(null);

}else{

try {

Thread.sleep(100);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

}).start();

}

}

补充：springmvc使用异步处理请求

同步请求图示：

同步处理的图示如上：HTTP请求，tomcat或其他中间件会有一个相应的线程来处理这个Http请求，所有的业务逻辑都会在这个线程里去执行，最后返回Http响应。但是tomcat等中间件，它们可以管理的线程数是有限的，当数量达到一定程度之后，再有请求进入，会被阻塞掉。

简单异步图示：

异步处理过程：当一个http请求进入后，tomcat等中间件的主线程调用副线程来执行业务逻辑，当副线程处理完成后，主线程再返回结果，在副线程处理整个业务逻辑的中，主线程会空闲出来去出来其他请求，也就是说采用上述这种模式处理http请求，服务器的吞吐量会有有明显的提升。使用异步返回，需使在web.xml将version配置为3.0版本的。

在servlet及所有的filter中配置异步支持。

简单实现如下：

更为复杂的业务场景的异步返回如下所示：

Htpp请求通过线程一处理，并将消息发送到消息队列，应用2处于不同的服务器，其接收到消息并将消息返回，线程2监听到处理结果，将消息返回，线程一及线程二不知道对方的存在。这种业务情况，单开一个线程是无法解决的，需要使用DeferredResult类。

简单的实现代码如下：

controller层：

@Controller

@RequestMapping("/test/")

@Slf4j

public class TestController {

@Autowired

private MockQueue mockQueue;

@Autowired

private DeferredResultHolder deferredResultHolder;

@RequestMapping("order")

@ResponseBody

public DeferredResult test() throws InterruptedException {

log.info("主线程开始");

String orderNo = RandomUtils.nextInt() + "";

mockQueue.setPlaceOrder(orderNo);

DeferredResult result = new DeferredResult();

deferredResultHolder.getMap().put(orderNo, result);

log.info("主线程结束");

return result;

}

}

伪消息队列类：

@Slf4j

@Component

public class MockQueue {

private String placeOrder;

private String compeleteOrder;

public String getPlaceOrder() {

return placeOrder;

}

public void setPlaceOrder(String placeOrder) throws InterruptedException {

new Thread(()->{ log.info("收到下单的请求");

this.placeOrder = placeOrder;

try {

Thread.sleep(1000);

} catch (InterruptedException e) {

e.printStackTrace();

}

this.compeleteOrder = placeOrder;

log.info("完成下单的请求");}).start();

}

public String getCompeleteOrder() {

return compeleteOrder;

}

public void setCompeleteOrder(String compeleteOrder) {

this.compeleteOrder = compeleteOrder;

}

}

伪队列监听类：

@Slf4j

@Component

public class QueueListener implements ApplicationListener{

@Autowihttp://red

private MockQueue mockQueue;

@Autowired

private DeferredResultHolder deferredResultHolder;

@Override

public void onApplicationEvent(ApplicationEvent applicationEvent) {

new Thread(() ->{

while (true){

if(StringUtils.isNotBlank(mockQueue.getCompeleteOrder())){

String orderNum = mockQueue.getCompeleteOrder();

log.info("返回订单处理结果" + orderNum);

deferredResultHolder.getMap().get(orderNum).setResult("success");

mockQueue.setCompeleteOrder(null);

}else {

try {

Thread.sleep(100);

} catch (InterruptedException e) {

e.printStackTrace();

}

UMzLV }

}

}).start();

}

}

容器类：

@Component

public class DeferredResultHolder {

private Map> map = new HashMap>();

public Map> getMap() {

return map;

}

public void setMap(Map> map) {

this.map = map;

}

}

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。