DeferredResult異步處理spring mvc Demo

一、概述

spring mvc同步接口在請求處理過程中一直處於阻塞狀態，而異步接口可以啟用后台線程去處理耗時任務。簡單來說適用場景：
1.高並發；
2.高IO耗時操作。

二、Demo

Spring MVC3.2之后支持異步請求，能夠在controller中返回一個Callable或者DeferredResult。
1.Callable實例

@Controller
public class CallableController {
    @RequestMapping(path = "/async1", method = RequestMethod.GET)
    @ResponseBody
    public Callable<String> asyncRequest() {
        return () -> {
            final long currentThread = Thread.currentThread().getId();
            final Date requestProcessingStarted = new Date();

            Thread.sleep(6000L);

            final Date requestProcessingFinished = new Date();

            return String.format(
                    "request: [threadId: %s, started: %s - finished: %s]"
                    , currentThread, requestProcessingStarted, requestProcessingFinished);
        };
    }
}

2.DeferredResult使用方式與Callable類似，但在返回結果上不一樣，它返回的時候實際結果可能沒有生成，實際的結果可能會在另外的線程里面設置到DeferredResult中去，能實現更加復雜的業務場景。

@Controller
public class DeferredResultController {
    private Map<Integer, DeferredResult<String>> deferredResultMap = new HashMap<>();

    @ResponseBody
    @GetMapping("/get")
    public DeferredResult<String> getId(@RequestParam Integer id) throws Exception {
        System.out.println("start hello");
        DeferredResult<String> deferredResult = new DeferredResult<>();

        //先存起來，等待觸發
        deferredResultMap.put(id, deferredResult);
        System.out.println("end hello");
        return deferredResult;
    }

    @ResponseBody
    @GetMapping("/set")
    public void setId(@RequestParam Integer id) throws Exception {
        // 讓所有hold住的請求給與響應
        if (deferredResultMap.containsKey(id)) {
            deferredResultMap.get(id).setResult("hello " + id);
        }
    }
}

當從瀏覽器請求http://localhost:8080/get/1時，頁面處於等待狀態；當訪問http://localhost:8080/set/1，前面的頁面會返回"hello 1"。

處理過程：

• controller 返回一個DeferredResult，我們把它保存到內存里或者List里面（供后續訪問）
• Spring MVC調用request.startAsync()，開啟異步處理
• 與此同時將DispatcherServlet里的攔截器、Filter等等都馬上退出主線程，但是response仍然保持打開的狀態
• 應用通過另外一個線程（可能是MQ消息、定時任務等）給DeferredResult set值。然后Spring MVC會把這個請求再次派發給servlet容器
• DispatcherServlet再次被調用，然后處理后續的標准流程

3.模擬場景：接口接收請求，推送到隊列receiveQueue，后台線程處理完成后推送到resultQueue，監聽器監聽resultQueue將結果賦值給DeferredResult，接口響應結果。
首先定義類Task:

public class Task<T> {
    private DeferredResult<String> result;
    private T message;
    private Boolean isTimeout;

定義MockQueue,用於管理隊列及處理數據：

@Component
public class MockQueue {
    /**
     * 接收隊列
     */
    private BlockingQueue<Task<String>> receiveQueue = new LinkedBlockingDeque<>(5000);
    /**
     * 結果隊列
     */
    private BlockingQueue<Task<String>> resultQueue = new LinkedBlockingDeque<>(5000);

    public MockQueue() {
        this.run();
    }

    /**
     * 接收task
     *
     * @param task task實體
     * @throws InterruptedException
     */
    public void put(Task<String> task) throws InterruptedException {
        receiveQueue.put(task);
    }

    /**
     * 獲取結果
     *
     * @return
     * @throws InterruptedException
     */
    public Task<String> get() throws InterruptedException {
        return resultQueue.take();
    }

    private void run() {
        new Thread(() -> {
            while (true) {
                try {
                    Task<String> task = receiveQueue.take();
                    System.out.println("receive data,start process!");
                    Thread.sleep(1000);
                    task.setMessage("success");

                    //任務超時，跳過
                    if (task.getIsTimeout()) {
                        continue;
                    }

                    resultQueue.put(task);
                    System.out.println("process done!");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

然后實現Controller異步接口：

@Controller
public class DeferredResultQueueController {
    @Autowired
    MockQueue queue;

    @ResponseBody
    @GetMapping("/test")
    public DeferredResult<String> test(@RequestParam Integer id) throws InterruptedException {
        System.out.println("start test");
        DeferredResult<String> deferredResult = new DeferredResult<>();
        Task<String> task = new Task<>(deferredResult, "任務", false);
        deferredResult.onTimeout(() -> {
            System.out.println("任務超時 id=" + id);
            task.setMessage("任務超時");
            task.setIsTimeout(true);
        });
        queue.put(task);
        return deferredResult;
    }
}

最后定義監聽器，將resultQueue的結果寫入DeferredResult。

@Component
public class QueueResultListener implements ApplicationListener<ContextRefreshedEvent> {
    @Autowired
    MockQueue mockQueue;

    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        new Thread(() -> {
            try {
                Task<String> task = mockQueue.get();
                task.getResult().setResult(task.getMessage());
                System.out.println("監聽器獲取到結果：task=" + task);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
    }
}

三、參考資料

https://www.baeldung.com/spring-deferred-result
https://cloud.tencent.com/developer/article/1497796
https://zhuanlan.zhihu.com/p/31223106