SpringMVC非同步處理操作(Callable和DeferredResult)
阿新 • • 發佈：2021-01-21
官方文件中說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();
}
}
補充：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<String> test() throws InterruptedException {
log.info("主執行緒開始");
String orderNo = RandomUtils.nextInt() + "";
mockQueue.setPlaceOrder(orderNo);
DeferredResult<String> result = new DeferredResult<String>();
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{
@Autowired
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();
}
}
}
}).start();
}
}
容器類：
@Component
public class DeferredResultHolder {
private Map<String,DeferredResult<String>>();
public Map<String,DeferredResult<String>> getMap() {
return map;
}
public void setMap(Map<String,DeferredResult<String>> map) {
this.map = map;
}
}
以上為個人經驗，希望能給大家一個參考，也希望大家多多支援我們。如有錯誤或未考慮完全的地方，望不吝賜教。