SpringMVC异步处理(Callable和DeferredResult)-蒲公英云

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

Callable

@GetMapping("/callable")
    public Callable<String> testCallable() throws InterruptedException {
        log.info("主线程开始！");
        Callable<String> result = new Callable<String>() {
            @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<String> order() throws Exception {
        logger.info("主线程开始");
        String orderNumber = RandomStringUtils.randomNumeric(8);
        mockQueue.setPlaceOrder(orderNumber);
        DeferredResult<String> 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<String, DeferredResult<String>> map = new HashMap<String, DeferredResult<String>>();
    public Map<String, DeferredResult<String>> getMap() {
        return map;
    }
    public void setMap(Map<String, DeferredResult<String>> 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<ContextRefreshedEvent> { 
    @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.isNotBlank(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(); } }