SpringBoot多線程進行異步請求的處理方式
SpringBoot多線程進行異步請求的處理
近期在協(xié)會博客園中，有人發(fā)布了博客，系統(tǒng)進行查重的時候由于機器最低配置進行大量計算時需要十秒左右時間才能處理完，由于一開始是單例模式，導致，在某人查重的時候整個系統(tǒng)是不會再響應別的請求的，導致了系統(tǒng)假死狀態(tài)，那么我們就應該使用多線程進行處理，將這種不能快速返回結(jié)果的方法交給線程池進行處理。
而我們自己使用Java Thread實現(xiàn)多線程又比較麻煩，在這里我們使用SpringBoot自帶的多線程支持，僅需兩步就可以對我們的查重方法利用多線程進行處理
第一步：編寫配置類
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;
@Configuration
@EnableAsync // 啟用異步任務
public class AsyncConfig {
// 聲明一個線程池(并指定線程池的名字)
@Bean("AsyncThread")
public Executor asyncExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
//核心線程數(shù)5：線程池創(chuàng)建時候初始化的線程數(shù)
executor.setCorePoolSize(5);
//最大線程數(shù)5：線程池最大的線程數(shù)，只有在緩沖隊列滿了之后才會申請超過核心線程數(shù)的線程
executor.setMaxPoolSize(10);
//緩沖隊列500：用來緩沖執(zhí)行任務的隊列
executor.setQueueCapacity(500);
//允許線程的空閑時間60秒：當超過了核心線程出之外的線程在空閑時間到達之后會被銷毀
executor.setKeepAliveSeconds(60);
//線程池名的前綴：設置好了之后可以方便我們定位處理任務所在的線程池
executor.setThreadNamePrefix("AsyncThread-");
executor.initialize();
return executor;
}
}
第二步：對方法使用注解標注為使用多線程進行處理
并將方法改寫返回類型（因為不能立即返回，所以需要將返回值改為CompletableFuture<String>,
返回的時候使用return CompletableFuture.completedFuture(str);）
方法示范：
@Async("AsyncThread")
@RequestMapping(value = "/addBlog")
public CompletableFuture<String> addBlog(HttpSession httpSession, Blog blog, String blogId, boolean tempSave) {
System.out.println("\n\n----------------------------------------------");
System.out.println(Thread.currentThread().getName() + "正在處理請求");
System.out.println("----------------------------------------------");
String result = "請求失敗";
//....你的業(yè)務邏輯
return CompletableFuture.completedFuture(result);
}
這樣以后你的這個方法將會交由線程池去進行處理，并將結(jié)果返回，一定要記得改返回值類型，否則返回的將是空的。
SpringBoot請求線程優(yōu)化
在我們的實際生產(chǎn)中，常常會遇到下面的這種情況，某個請求非常耗時(大約5s返回)，當大量的訪問該請求的時候，再請求其他服務時，會造成沒有連接使用的情況，造成這種現(xiàn)象的主要原因是，我們的容器(tomcat)中線程的數(shù)量是一定的，例如500個，當這500個線程都用來請求服務的時候，再有請求進來，就沒有多余的連接可用了，只能拒絕連接。要是我們在請求耗時服務的時候，能夠異步請求(請求到controller中時，則容器線程直接返回，然后使用系統(tǒng)內(nèi)部的線程來執(zhí)行耗時的服務，等到服務有返回的時候，再將請求返回給客戶端)，那么系統(tǒng)的吞吐量就會得到很大程度的提升了。當然，大家可以直接使用Hystrix的資源隔離來實現(xiàn)，今天我們的重點是spring mvc是怎么來實現(xiàn)這種異步請求的。
使用Callable來實現(xiàn)
controller如下：
@RestController
public class HelloController {undefined
private static final Logger logger = LoggerFactory.getLogger(HelloController.class);
@Autowired
private HelloService hello;
@GetMapping("/helloworld")
public String helloWorldController() {undefined
return hello.sayHello();
}
/**
* 異步調(diào)用restful
* 當controller返回值是Callable的時候，springmvc就會啟動一個線程將Callable交給TaskExecutor去處理
* 然后DispatcherServlet還有所有的spring攔截器都退出主線程，然后把response保持打開的狀態(tài)
* 當Callable執(zhí)行結(jié)束之后，springmvc就會重新啟動分配一個request請求，然后DispatcherServlet就重新
* 調(diào)用和處理Callable異步執(zhí)行的返回結(jié)果， 然后返回視圖
*
* @return
*/
@GetMapping("/hello")
public Callable<String> helloController() {undefined
logger.info(Thread.currentThread().getName() + " 進入helloController方法");
Callable<String> callable = new Callable<String>() {undefined
@Override
public String call() throws Exception {undefined
logger.info(Thread.currentThread().getName() + " 進入call方法");
String say = hello.sayHello();
logger.info(Thread.currentThread().getName() + " 從helloService方法返回");
return say;
}
};
logger.info(Thread.currentThread().getName() + " 從helloController方法返回");
return callable;
}
}
我們首先來看下上面這兩個請求的區(qū)別：
下面這個是沒有使用異步請求的
2017-12-07 18:05:42.351 INFO 3020 --- [nio-8060-exec-5] c.travelsky.controller.HelloController : http-nio-8060-exec-5 進入helloWorldController方法
2017-12-07 18:05:42.351 INFO 3020 --- [nio-8060-exec-5] com.travelsky.service.HelloService : http-nio-8060-exec-5 進入sayHello方法!
2017-12-07 18:05:44.351 INFO 3020 --- [nio-8060-exec-5] c.travelsky.controller.HelloController : http-nio-8060-exec-5 從helloWorldController方法返回
我們可以看到，請求從頭到尾都只有一個線程，并且整個請求耗費了2s鐘的時間。
下面，我們再來看下使用Callable異步請求的結(jié)果：
2017-12-07 18:11:55.671 INFO 6196 --- [nio-8060-exec-1] c.travelsky.controller.HelloController : http-nio-8060-exec-1 進入helloController方法
2017-12-07 18:11:55.672 INFO 6196 --- [nio-8060-exec-1] c.travelsky.controller.HelloController : http-nio-8060-exec-1 從helloController方法返回
2017-12-07 18:11:55.676 INFO 6196 --- [nio-8060-exec-1] c.t.i.MyAsyncHandlerInterceptor : http-nio-8060-exec-1 進入afterConcurrentHandlingStarted方法
2017-12-07 18:11:55.676 INFO 6196 --- [ MvcAsync1] c.travelsky.controller.HelloController : MvcAsync1 進入call方法
2017-12-07 18:11:55.676 INFO 6196 --- [ MvcAsync1] com.travelsky.service.HelloService : MvcAsync1 進入sayHello方法!
2017-12-07 18:11:57.677 INFO 6196 --- [ MvcAsync1] c.travelsky.controller.HelloController : MvcAsync1 從helloService方法返回
2017-12-07 18:11:57.721 INFO 6196 --- [nio-8060-exec-2] c.t.i.MyAsyncHandlerInterceptor : http-nio-8060-exec-2服務調(diào)用完成，返回結(jié)果給客戶端
從上面的結(jié)果中，我們可以看出，容器的線程http-nio-8060-exec-1這個線程進入controller之后，就立即返回了，具體的服務調(diào)用是通過MvcAsync2這個線程來做的，當服務執(zhí)行完要返回后，容器會再啟一個新的線程http-nio-8060-exec-2來將結(jié)果返回給客戶端或瀏覽器，整個過程response都是打開的，當有返回的時候，再從server端推到response中去。
1、異步調(diào)用的另一種方式
上面的示例是通過callable來實現(xiàn)的異步調(diào)用，其實還可以通過WebAsyncTask，也能實現(xiàn)異步調(diào)用，下面看示例：
@RestController
public class HelloController {undefined
private static final Logger logger = LoggerFactory.getLogger(HelloController.class);
@Autowired
private HelloService hello;
/**
* 帶超時時間的異步請求 通過WebAsyncTask自定義客戶端超時間
*
* @return
*/
@GetMapping("/world")
public WebAsyncTask<String> worldController() {undefined
logger.info(Thread.currentThread().getName() + " 進入helloController方法");
// 3s鐘沒返回，則認為超時
WebAsyncTask<String> webAsyncTask = new WebAsyncTask<>(3000, new Callable<String>() {undefined
@Override
public String call() throws Exception {undefined
logger.info(Thread.currentThread().getName() + " 進入call方法");
String say = hello.sayHello();
logger.info(Thread.currentThread().getName() + " 從helloService方法返回");
return say;
}
});
logger.info(Thread.currentThread().getName() + " 從helloController方法返回");
webAsyncTask.onCompletion(new Runnable() {undefined
@Override
public void run() {undefined
logger.info(Thread.currentThread().getName() + " 執(zhí)行完畢");
}
});
webAsyncTask.onTimeout(new Callable<String>() {undefined
@Override
public String call() throws Exception {undefined
logger.info(Thread.currentThread().getName() + " onTimeout");
// 超時的時候，直接拋異常，讓外層統(tǒng)一處理超時異常
throw new TimeoutException("調(diào)用超時");
}
});
return webAsyncTask;
}
/**
* 異步調(diào)用，異常處理，詳細的處理流程見MyExceptionHandler類
*
* @return
*/
@GetMapping("/exception")
public WebAsyncTask<String> exceptionController() {undefined
logger.info(Thread.currentThread().getName() + " 進入helloController方法");
Callable<String> callable = new Callable<String>() {undefined
@Override
public String call() throws Exception {undefined
logger.info(Thread.currentThread().getName() + " 進入call方法");
throw new TimeoutException("調(diào)用超時!");
}
};
logger.info(Thread.currentThread().getName() + " 從helloController方法返回");
return new WebAsyncTask<>(20000, callable);
}
}
運行結(jié)果如下：
2017-12-07 19:10:26.582 INFO 6196 --- [nio-8060-exec-4] c.travelsky.controller.HelloController : http-nio-8060-exec-4 進入helloController方法
2017-12-07 19:10:26.585 INFO 6196 --- [nio-8060-exec-4] c.travelsky.controller.HelloController : http-nio-8060-exec-4 從helloController方法返回
2017-12-07 19:10:26.589 INFO 6196 --- [nio-8060-exec-4] c.t.i.MyAsyncHandlerInterceptor : http-nio-8060-exec-4 進入afterConcurrentHandlingStarted方法
2017-12-07 19:10:26.591 INFO 6196 --- [ MvcAsync2] c.travelsky.controller.HelloController : MvcAsync2 進入call方法
2017-12-07 19:10:26.591 INFO 6196 --- [ MvcAsync2] com.travelsky.service.HelloService : MvcAsync2 進入sayHello方法!
2017-12-07 19:10:28.591 INFO 6196 --- [ MvcAsync2] c.travelsky.controller.HelloController : MvcAsync2 從helloService方法返回
2017-12-07 19:10:28.600 INFO 6196 --- [nio-8060-exec-5] c.t.i.MyAsyncHandlerInterceptor : http-nio-8060-exec-5服務調(diào)用完成，返回結(jié)果給客戶端
2017-12-07 19:10:28.601 INFO 6196 --- [nio-8060-exec-5] c.travelsky.controller.HelloController : http-nio-8060-exec-5 執(zhí)行完畢
這種方式和上面的callable方式最大的區(qū)別就是，WebAsyncTask支持超時，并且還提供了兩個回調(diào)函數(shù)，分別是onCompletion和onTimeout，顧名思義，這兩個回調(diào)函數(shù)分別在執(zhí)行完成和超時的時候回調(diào)。
2、Deferred方式實現(xiàn)異步調(diào)用
在我們是生產(chǎn)中，往往會遇到這樣的情景，controller中調(diào)用的方法很多都是和第三方有關(guān)的，例如JMS，定時任務，隊列等，拿JMS來說，比如controller里面的服務需要從JMS中拿到返回值，才能給客戶端返回，而從JMS拿值這個過程也是異步的，這個時候，我們就可以通過Deferred來實現(xiàn)整個的異步調(diào)用。
首先，我們來模擬一個長時間調(diào)用的任務，代碼如下：
@Component
public class LongTimeTask {undefined
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@Async
public void execute(DeferredResult<String> deferred){undefined
logger.info(Thread.currentThread().getName() + "進入 taskService 的 execute方法");
try {undefined
// 模擬長時間任務調(diào)用，睡眠2s
TimeUnit.SECONDS.sleep(2);
// 2s后給Deferred發(fā)送成功消息，告訴Deferred，我這邊已經(jīng)處理完了，可以返回給客戶端了
deferred.setResult("world");
} catch (InterruptedException e) {undefined
e.printStackTrace();
}
}
}
接著，我們就來實現(xiàn)異步調(diào)用，controller如下：
@RestController
public class AsyncDeferredController {undefined
private final Logger logger = LoggerFactory.getLogger(this.getClass());
private final LongTimeTask taskService;
@Autowired
public AsyncDeferredController(LongTimeTask taskService) {undefined
this.taskService = taskService;
}
@GetMapping("/deferred")
public DeferredResult<String> executeSlowTask() {undefined
logger.info(Thread.currentThread().getName() + "進入executeSlowTask方法");
DeferredResult<String> deferredResult = new DeferredResult<>();
// 調(diào)用長時間執(zhí)行任務
taskService.execute(deferredResult);
// 當長時間任務中使用deferred.setResult("world");這個方法時，會從長時間任務中返回，繼續(xù)controller里面的流程
logger.info(Thread.currentThread().getName() + "從executeSlowTask方法返回");
// 超時的回調(diào)方法
deferredResult.onTimeout(new Runnable(){undefined
@Override
public void run() {undefined
logger.info(Thread.currentThread().getName() + " onTimeout");
// 返回超時信息
deferredResult.setErrorResult("time out!");
}
});
// 處理完成的回調(diào)方法，無論是超時還是處理成功，都會進入這個回調(diào)方法
deferredResult.onCompletion(new Runnable(){undefined
@Override
public void run() {undefined
logger.info(Thread.currentThread().getName() + " onCompletion");
}
});
return deferredResult;
}
}
執(zhí)行結(jié)果如下：
2017-12-07 19:25:40.192 INFO 6196 --- [nio-8060-exec-7] c.t.controller.AsyncDeferredController : http-nio-8060-exec-7進入executeSlowTask方法
2017-12-07 19:25:40.193 INFO 6196 --- [nio-8060-exec-7] .s.a.AnnotationAsyncExecutionInterceptor : No TaskExecutor bean found for async processing
2017-12-07 19:25:40.194 INFO 6196 --- [nio-8060-exec-7] c.t.controller.AsyncDeferredController : http-nio-8060-exec-7從executeSlowTask方法返回
2017-12-07 19:25:40.198 INFO 6196 --- [nio-8060-exec-7] c.t.i.MyAsyncHandlerInterceptor : http-nio-8060-exec-7 進入afterConcurrentHandlingStarted方法
2017-12-07 19:25:40.202 INFO 6196 --- [cTaskExecutor-1] com.travelsky.controller.LongTimeTask : SimpleAsyncTaskExecutor-1進入 taskService 的 execute方法
2017-12-07 19:25:42.212 INFO 6196 --- [nio-8060-exec-8] c.t.i.MyAsyncHandlerInterceptor : http-nio-8060-exec-8服務調(diào)用完成，返回結(jié)果給客戶端
2017-12-07 19:25:42.213 INFO 6196 --- [nio-8060-exec-8] c.t.controller.AsyncDeferredController : http-nio-8060-exec-8 onCompletion
從上面的執(zhí)行結(jié)果不難看出，容器線程會立刻返回，應用程序使用線程池里面的cTaskExecutor-1線程來完成長時間任務的調(diào)用，當調(diào)用完成后，容器又啟了一個連接線程，來返回最終的執(zhí)行結(jié)果。
這種異步調(diào)用，在容器線程資源非常寶貴的時候，能夠大大的提高整個系統(tǒng)的吞吐量。
ps：異步調(diào)用可以使用AsyncHandlerInterceptor進行攔截，使用示例如下：
@Component
public class MyAsyncHandlerInterceptor implements AsyncHandlerInterceptor {undefined
private static final Logger logger = LoggerFactory.getLogger(MyAsyncHandlerInterceptor.class);
@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler)
throws Exception {undefined
return true;
}
@Override
public void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler,
ModelAndView modelAndView) throws Exception {undefined
// HandlerMethod handlerMethod = (HandlerMethod) handler;
logger.info(Thread.currentThread().getName()+ "服務調(diào)用完成，返回結(jié)果給客戶端");
}
@Override
public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex)
throws Exception {undefined
if(null != ex){undefined
System.out.println("發(fā)生異常:"+ex.getMessage());
}
}
@Override
public void afterConcurrentHandlingStarted(HttpServletRequest request, HttpServletResponse response, Object handler)
throws Exception {undefined
// 攔截之后，重新寫回數(shù)據(jù)，將原來的hello world換成如下字符串
String resp = "my name is chhliu!";
response.setContentLength(resp.length());
response.getOutputStream().write(resp.getBytes());
logger.info(Thread.currentThread().getName() + " 進入afterConcurrentHandlingStarted方法");
}
}
有興趣的可以了解下，我這里的主題是異步調(diào)用，其他的相關(guān)知識點，以后在講解吧。希望能給大家一個參考，也希望大家多多支持腳本之家。
相關(guān)文章
SpringBoot前后端交互、全局異常處理之后端異常信息拋到前端顯示彈窗
Spring Boot是一個用于構(gòu)建獨立的、基于生產(chǎn)級別的Spring應用程序的框架,下面這篇文章主要給大家介紹了關(guān)于SpringBoot前后端交互、全局異常處理之后端異常信息拋到前端顯示彈窗的相關(guān)資料,需要的朋友可以參考下2024-08-08
本文主要介紹了MySQL的簡易封裝以及使用。具有一定的參考價值，下面跟著小編一起來看下吧2017-01-01
Spring中自帶的@Schedule實現(xiàn)自動任務的過程解析
這篇文章主要介紹了關(guān)于Spring中自帶的@Schedule實現(xiàn)自動任務,本文通過實例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2023-06-06
這篇文章主要介紹了Java中@InitBinder注解的使用原理詳解,在Java中,@InitBinder 注解是用于標記一個方法,該方法將在控制器(Controller)處理請求之前被調(diào)用,用于進行一些數(shù)據(jù)綁定和預處理的操作,需要的朋友可以參考下2023-10-10
這篇文章主要介紹了使用@Valid 校驗嵌套對象方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-10-10
這篇文章主要介紹了jdk在centos中安裝配置圖文教程,文中給出大家jdk下載地址，需要的朋友可以參考下2018-04-04
這篇文章主要介紹了SpringBoot加載應用事件監(jiān)聽器代碼實例,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-06-06
springboot如何使用redis的incr創(chuàng)建分布式自增id
這篇文章主要介紹了springboot如何使用redis的incr創(chuàng)建分布式自增id，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-11-11
在本文中小編給大家整理了關(guān)于java數(shù)組的初始化及操作的相關(guān)知識點內(nèi)容，需要的讀者們參考下。2019-07-07
這篇文章主要介紹了Java 淺復制和深復制的實例詳解的相關(guān)資料,這里提供實例幫助大家學習理解這部分內(nèi)容，需要的朋友可以參考下2017-08-08