Springboot詳解線程池與多線程及阻塞隊列的應用詳解
更新時間：2022年06月17日 10:56:01 作者：zhangbeizhen18
本例應用線程池、多線程、阻塞隊列處理一個流程任務。本例處理一個訂單流程，主要包括生成訂單、訂單處理、訂單入庫，下面我們一起看看
版本：Spring Boot 2.6.3
一、案例場景
1>web端接收restful請求生成任務A，并把任務放入隊列Queue_A。
2>線程池A的任務線程從隊列Queue_A取出任務，處理完成后放入Queue_B。
3>線程池B的任務線程從Queue_B取出任務，處理完成后入庫。
本例就使用兩個任務步驟，按需擴展延長任務鏈。
二、使用類
java.util.LinkedHashMap，雙向鏈表。
java.util.concurrent.BlockingQueue，阻塞隊列接口。
java.util.concurrent.LinkedBlockingQueue，阻塞隊列實現(xiàn)類。
java.util.concurrent.CountDownLatch，線程計數器。
java.util.concurrent.locks.ReentrantLock，可重入鎖。
三、本例說明
1.接收web請求
OrderController接收web請求，業(yè)務數據封裝成任務對象，并寫入隊列QUEUE_A。Web請求結束，立即返回。
2.后臺任務處理
FlowStarter流程啟動器
管理FlowManager，創(chuàng)建流程管理器和啟動流程管理器。創(chuàng)建線程池容器StepContainer，指定隊列、線程池線程數量，以及業(yè)務處理Handler。
FlowManager流程管理器
管理線程池容器StepContainer。創(chuàng)建線程池容器，啟動線程池容器，關閉線程池容器，線程池容器之間數據傳遞。使用LinkedHashMap維護一個流程中的多個線程池容器。
StepContainer線程池容器
創(chuàng)建線程池，啟動線程執(zhí)行器(Executor)，初始化業(yè)務處理Handler，讀寫隊列。使用LinkedHashMap維護一個流程中的多個StepExecutor。
StepExecutor線程執(zhí)行器
執(zhí)行抽象公用業(yè)務邏輯。實現(xiàn)線程Runnable接口。調用StepHandler的實現(xiàn)類的execute執(zhí)行具體業(yè)務邏輯。
StepHandler業(yè)務處理器handler
具體業(yè)務在StepHandler的實現(xiàn)類的execute中實現(xiàn)。
任務模型對象StepModel和執(zhí)行結果對象StepResult
每個具體業(yè)務數據必須包裝成任務模型對象StepModel，執(zhí)行結果包裝成執(zhí)行結果對象StepResult，才能在線程池和隊列中流轉。
3.關系說明
一個FlowStarter可以啟動一個或者多個FlowManager。支持一對多和一對一，按需擴展。
一個FlowManager對應一個業(yè)務流程。一個業(yè)務流程可以拆分為多個步驟。一個步驟對應一個線程池容器StepContainer。一個線程池容器StepContainer，啟動多個線程執(zhí)行器StepExecutor。效果就是并發(fā)執(zhí)行任務。
一個業(yè)務流程拆分成若干個步驟，每個步驟之間數據流轉，使用任務模型StepModel中的狀態(tài)標識isFinished，isPutInQueueAgain，isPutInQueueNext 字段來分析任務流向。使用StepModel的StepResult的 nextStepName字段來識別具體流向的線程池容器。
四、代碼
1.OrderController
OrderController，接收請求、封裝任務、寫隊列。
@Slf4j
@RestController
@RequestMapping("/order")
public class OrderController {
@PostMapping("/f1")
public Object f1(@RequestBody Object obj) {
log.info("OrderController->f1,接收參數,obj = " + obj.toString());
Map objMap = (Map) obj;
OrderInfo orderInfo = new OrderInfo();
orderInfo.setUserName((String) objMap.get("userName"));
orderInfo.setTradeName((String) objMap.get("tradeName"));
orderInfo.setOrderTime(System.currentTimeMillis());
LinkedBlockingQueue<StepModel> queueA = FlowQueue.getBlockingQueue("QUEUE_A");
QueueUtils.putStepPutInQueue(queueA,orderInfo);
log.info("OrderController->f1,返回." );
return ResultObj.builder().code("200").message("成功").build();
}
}2.FlowStarter流程啟動器
FlowStarter，后臺任務線程池和線程啟動。實現(xiàn)InitializingBean了接口。那么在spring初始化化bean完成后，就能觸發(fā)啟動線程池和線程。
@Slf4j
@Service
public class FlowStarter implements InitializingBean {
@Override
public void afterPropertiesSet() throws Exception {
log.info("FlowWorker創(chuàng)建流程.");
FlowManager flowManager = new FlowManager();
flowManager.buildContainer(ConstantUtils.STEP_01,5,
FlowQueue.getBlockingQueue("QUEUE_A"), Step01Handler.class
);
flowManager.buildContainer(ConstantUtils.STEP_02,5,
FlowQueue.getBlockingQueue("QUEUE_B"), Step02Handler.class
);
flowManager.startContainers();
log.info("FlowWorker啟動流程完成.");
}
}3.FlowManager流程管理器
一個FlowManager流程管理器，維護多個線程池容器StepContainer，共同完成一個流程的多個步驟。
public class FlowManager {
// 管理器名稱
private String name;
// 管理線程池容器
private Map<String, StepContainer> stepContainerMap = new LinkedHashMap<>();
public FlowManager() {}
// 創(chuàng)建線程池容器
public void buildContainer(String name, int poolSize, BlockingQueue<StepModel> queue,
Class<? extends StepHandler> handlerClazz) {
StepContainer stepWorker = new StepContainer();
stepWorker.createThreadPool(poolSize, queue, handlerClazz);
stepWorker.setName(name);
stepWorker.setFlowManager(this);
this.stepContainerMap.put(name, stepWorker);
}
// 啟動線程池容器
public void startContainers() {
for (StepContainer stepContainer : this.stepContainerMap.values()) {
stepContainer.startRunExecutor();
}
}
// 關閉線程池容器
public void stopContainers() {
for (StepContainer stepContainer : this.stepContainerMap.values()) {
stepContainer.stopRunExecutor();
}
this.stepContainerMap.clear();
}
// 任務放入下一個線程池
public boolean sendToNextContainer(String nextStepName, Object obj) {
if (nextStepName != null && !StringUtils.equals(nextStepName, "")) {
if (this.stepContainerMap.containsKey(nextStepName)) {
this.stepContainerMap.get(nextStepName).putStepInQueue(obj);
return true;
} else {
return false;
}
} else {
return false;
}
}
public String getName() {
return name;
}
}4.StepContainer線程池容器
StepContainer線程池容器，維護多個線程執(zhí)行器StepExecutor，實現(xiàn)多線程異步完成每個獨立任務。
@Slf4j
public class StepContainer {
// 線程池名稱
private String name;
// 線程池
private ExecutorService threadPool;
// 線程數目
private int nThreads = 0;
// 線程處理業(yè)務handler類
private Class handlerClazz;
// 線程處理業(yè)務隊列
private BlockingQueue<StepModel> queue = null;
// 線程池內線程管理
private Map<String, StepExecutor> stepExecutorMap = new LinkedHashMap<>();
// 線程池運行狀態(tài)
private boolean isRun = false;
// 線程池管理器
private FlowManager flowManager = null;
// 構造函數
public StepContainer() {}
// 創(chuàng)建線程池
public boolean createThreadPool(int nThreads, BlockingQueue<StepModel> queue,
Class<? extends StepHandler> handlerClazz) {
try {
this.nThreads = nThreads;
this.queue = queue;
this.handlerClazz = handlerClazz;
this.threadPool = Executors.newFixedThreadPool(this.nThreads, new ThreadFactory() {
@Override
public Thread newThread(Runnable runnable) {
return new Thread(runnable);
}
});
} catch (Exception e) {
e.printStackTrace();
return false;
}
return true;
}
// 啟動線程
public void startRunExecutor() {
if (!this.isRun) {
if (this.handlerClazz != null) {
log.info("線程池: " + this.name + ",啟動,加載線程Executor.");
StepExecutor stepExecutor;
String executorName = "";
for (int num = 0; num < this.nThreads; num++) {
try {
executorName = this.name + "_" + (num + 1);
StepHandler stepHandler = (StepHandler) createStepHandler(this.handlerClazz);
stepExecutor = new StepExecutor(executorName, this.queue, stepHandler, this);
this.threadPool.execute(stepExecutor);
this.stepExecutorMap.put(executorName, stepExecutor);
} catch (Exception e) {
e.printStackTrace();
}
}
this.isRun = true;
}
}
}
// 關閉線程
public void stopRunExecutor() {
if (isRun) {
Iterator iterator = this.stepExecutorMap.values().iterator();
while (iterator.hasNext()) {
StepExecutor stepExecutor = (StepExecutor) iterator.next();
stepExecutor.stop();
}
this.stepExecutorMap.clear();
this.isRun = false;
}
}
// 從隊列獲取任務
public StepModel getStepFromQueue() {
StepModel stepModel = null;
synchronized (this.queue) {
try {
if (this.queue.size() > 0) {
stepModel = this.queue.take();
}
} catch (Exception e) {
log.info("從隊列獲取任務異常.");
e.printStackTrace();
}
}
return stepModel;
}
// 任務放入隊列
public void putStepInQueue(Object obj) {
try {
StepModel stepModel = new StepModel(obj);
stepModel.setPutInQueueTime(System.currentTimeMillis());
this.queue.put(stepModel);
} catch (InterruptedException e) {
log.info("任務放入隊列異常.");
e.printStackTrace();
}
}
// 重新放入
public void putStepInQueueAgain(StepModel stepModel) {
stepModel.setFinished(false);
stepModel.setPutInQueueNext(false);
stepModel.setPutInQueueAgain(false);
try {
this.queue.put(stepModel);
} catch (InterruptedException e) {
log.info("任務重新放入隊列異常.");
e.printStackTrace();
}
}
// 清空隊列
public void clearQueue() {
if (this.queue != null) {
this.queue.clear();
}
}
// 初始化實例對象
public Object createStepHandler(Class clazz)
throws InstantiationException, IllegalAccessException {
Object object = clazz.newInstance();
return object;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public FlowManager getFlowManager() {
return flowManager;
}
public void setFlowManager(FlowManager flowManager) {
this.flowManager = flowManager;
}
}5.StepExecutor線程執(zhí)行器
StepExecutor線程執(zhí)行器，實現(xiàn)Runnable接口。線程執(zhí)行單元通用邏輯，具體業(yè)務邏輯通過調用StepHandler的execute方法實現(xiàn)。
@Slf4j
public class StepExecutor implements Runnable {
// 執(zhí)行器名稱
private String name;
// 線程執(zhí)行的任務
private StepModel stepModel;
// 線程執(zhí)行的隊列
private BlockingQueue<StepModel> queue;
// 線程執(zhí)行的業(yè)務處理邏輯
private Object stepHandler;
// 線程運行狀態(tài)
private volatile boolean isRun = false;
// 線程開啟(True)和關閉(False)
private volatile boolean isClose = false;
// 線程隸屬容器
private StepContainer stepContainer;
// 線程計數器(關閉線程使用)
private CountDownLatch countDownLatch = null;
public StepExecutor() {}
public StepExecutor(String name, BlockingQueue<StepModel> queue,
StepHandler stepHandler, StepContainer stepContainer) {
this.name = name;
this.queue = queue;
this.stepHandler = stepHandler;
this.stepContainer = stepContainer;
}
@Override
public void run() {
this.isRun = true;
this.countDownLatch = new CountDownLatch(1);
// 沒收到關閉信號,則循環(huán)運行
while (!this.isClose) {
this.stepModel = null;
String threadName = "【線程池:" + this.stepContainer.getName()
+ ",線程:" + Thread.currentThread().getName() + "】";
// 循環(huán)運行,為防止中斷和卡主,需捕獲異常
try {
StepHandler stepHandler = (StepHandler) this.stepHandler;
this.stepModel = this.stepContainer.getStepFromQueue();
if (this.stepModel != null) {
log.info(threadName + ",處理任務.");
this.stepModel.getStepResultList().clear();
stepHandler.execute(this.stepModel);
// 執(zhí)行完成后結果數據
List<StepResult> stepResultList = this.stepModel.getStepResultList();
boolean isFinished = this.stepModel.isFinished();
boolean isPutInQueueAgain = this.stepModel.isPutInQueueAgain();
boolean isPutInQueueNext = this.stepModel.isPutInQueueNext();
if (isFinished && !isPutInQueueAgain && !isPutInQueueNext) {
log.info(threadName + ",任務結束.");
}
if (!isFinished && isPutInQueueAgain && !isPutInQueueNext) {
log.info(threadName + ",任務在本步驟未完成,重新放隊列.");
this.stepContainer.putStepInQueueAgain(this.stepModel);
}
if (!isFinished && !isPutInQueueAgain && isPutInQueueNext) {
int resultNum = stepResultList.size();
if (resultNum > 0) {
for (StepResult stepResult : stepResultList) {
log.info(threadName + ",任務在本步驟已經完成,發(fā)送給下一個線程池: "
+ stepResult.getNextStepName() + ",執(zhí)行.");
this.stepContainer.getFlowManager().sendToNextContainer(
stepResult.getNextStepName(),
stepResult.getResult());
}
}
}
} else {
threadToSleep(1000 * 3L);
}
} catch (Exception e) {
log.info("執(zhí)行器異常.");
e.printStackTrace();
this.stepContainer.putStepInQueueAgain(this.stepModel);
}
}
// 跳出循環(huán)后,線程計數減1
this.countDownLatch.countDown();
this.isRun = false;
}
public void stop() {
this.isClose = true;
if (this.countDownLatch != null) {
while (this.countDownLatch.getCount() > 0L) {
try {
this.countDownLatch.await();
} catch (InterruptedException e) {
log.info("線程關閉異常.");
e.printStackTrace();
}
}
}
this.isClose = false;
}
public void threadToSleep(long time) {
try {
Thread.sleep(time);
} catch (Exception e) {
log.info("線程休眠異常.");
e.printStackTrace();
}
}
}6.StepHandler業(yè)務處理handler
StepHandler是StepExecutor線程執(zhí)行器，具體執(zhí)行業(yè)務邏輯的入口。
StepHandler抽象類
每個具體的實現(xiàn)類都繼承抽象的StepHandler。
public abstract class StepHandler {
public StepHandler() {}
public abstract void execute(StepModel stepModel);
}Step01Handler
Step01Handler是StepHandler實現(xiàn)類，從隊列中取任務執(zhí)行，執(zhí)行完成后放入下一個業(yè)務處理器Step02Handler。
@Slf4j
public class Step01Handler extends StepHandler {
@Override
public void execute(StepModel stepModel) {
log.info("Step01Handler執(zhí)行開始,stepModel: " + stepModel.toString());
OrderInfo orderInfo = (OrderInfo) stepModel.getObj();
List<StepResult> stepResultList = stepModel.getStepResultList();
try {
log.info("Step01Handler執(zhí)行,處理訂單.");
String orderNo = UUID.randomUUID().toString()
.replace("-", "").toUpperCase();
orderInfo.setOrderNo(orderNo);
orderInfo.setPlatformType("線上");
orderInfo.setOrderSource("Web");
stepModel.setFinished(false);
stepModel.setPutInQueueNext(true);
stepModel.setPutInQueueAgain(false);
stepResultList.add(new StepResult(ConstantUtils.STEP_02, orderInfo));
} catch (Exception e) {
stepModel.setFinished(false);
stepModel.setPutInQueueNext(false);
stepModel.setPutInQueueAgain(true);
stepResultList.add(new StepResult(ConstantUtils.STEP_01, orderInfo));
}
log.info("Step01Handler執(zhí)行完成,stepModel: " + stepModel.toString());
}
}Step02Handler
Step02Handler是StepHandler實現(xiàn)類，從隊列中取任務執(zhí)行。
@Slf4j
public class Step02Handler extends StepHandler{
@Override
public void execute(StepModel stepModel) {
log.info("Step02Handler執(zhí)行開始,stepModel: " + stepModel.toString());
OrderInfo orderInfo = (OrderInfo) stepModel.getObj();
List<StepResult> stepResultList = stepModel.getStepResultList();
try {
orderInfo.setEndTime(System.currentTimeMillis());
stepModel.setFinished(true);
stepModel.setPutInQueueNext(false);
stepModel.setPutInQueueAgain(false);
log.info("Step02Handler執(zhí)行,入庫.");
} catch (Exception e) {
stepModel.setFinished(true);
stepModel.setPutInQueueNext(false);
stepModel.setPutInQueueAgain(false);
}
log.info("Step02Handler執(zhí)行完成,stepModel: " + stepModel.toString());
}
}7.阻塞隊列
BlockingQueue是線程安全的阻塞隊列。
7.1 FlowQueue
FlowQueue，管理本例使用的兩個阻塞隊列。
public class FlowQueue {
private static final LinkedBlockingQueue<StepModel> queueA = new LinkedBlockingQueue<StepModel>();
private static final LinkedBlockingQueue<StepModel> queueB = new LinkedBlockingQueue<StepModel>();
public static LinkedBlockingQueue<StepModel> getBlockingQueue(String queueName) {
LinkedBlockingQueue<StepModel> queue = null;
switch (queueName) {
case "QUEUE_A":
queue = queueA;
break;
case "QUEUE_B":
queue = queueB;
break;
}
return queue;
}
}7.2 QueueUtils
QueueUtils，隊列簡易工具。
@Slf4j
public class QueueUtils {
public static StepModel getStepFromQueue(
LinkedBlockingQueue<StepModel> queue) {
StepModel stepModel = null;
try {
if (queue.size() > 0) {
stepModel = queue.take();
}
} catch (Exception e) {
log.info("讀隊列異常.");
e.printStackTrace();
}
return stepModel;
}
public static void putStepPutInQueue(
LinkedBlockingQueue<StepModel> queue, Object obj) {
try {
StepModel stepModel = new StepModel(obj);
stepModel.setPutInQueueTime(System.currentTimeMillis());
queue.put(stepModel);
} catch (Exception e) {
log.info("寫隊列異常.");
e.printStackTrace();
}
}
public static int getQueueSize(
LinkedBlockingQueue<StepModel> queue) {
int size = 0;
try {
size = queue.size();
} catch (Exception e) {
log.info("獲取隊列Size異常.");
e.printStackTrace();
}
return size;
}
}7.3 ConstantUtils
ConstantUtils，管理常量，即線程池名稱。
public class ConstantUtils {
public static final String STEP_01 = "STEP_01_THREAD_POOL";
public static final String STEP_02 = "STEP_02_THREAD_POOL";
}8.任務模型
任務模型，即具體需要處理對象，封裝成線程使用的任務模型，這樣可以把業(yè)務和流程框架解耦。
8.1 StepModel
StepModel，任務模型封裝。
@Data
public class StepModel {
// 任務對象
private Object obj;
// 任務執(zhí)行結果
private List<StepResult> stepResultList;
// 任務接收時間
private long putInQueueTime;
// 任務完成標識
private boolean isFinished = false;
// 任務重新放入隊列標識
private boolean isPutInQueueAgain = false;
// 任務放入下一個隊列標識
private boolean isPutInQueueNext = false;
public StepModel(Object object) {
this.obj = object;
this.stepResultList = new ArrayList<>();
}
}8.2 StepResult
StepResult，執(zhí)行結果模型封裝。
@Data
public class StepResult {
// 目標線程池名
private String nextStepName;
// 執(zhí)行結果
private Object result;
public StepResult(String nextStepName,Object result){
this.nextStepName = nextStepName;
this.result = result;
}
}9.業(yè)務數據模型
業(yè)務數據模型，即生成具體需要處理的數據，在傳入給線程池的線程執(zhí)行前，需要封裝成任務模型。
9.1 OrderInfo
OrderInfo，本例要處理的業(yè)務數據模型。
@Data
@NoArgsConstructor
public class OrderInfo {
private String userName;
private String orderNo;
private String tradeName;
private String platformType;
private String orderSource;
private long orderTime;
private long endTime;
}9.2 ResultObj
ResultObj，web請求返回的統(tǒng)一封裝對象。
@Data
@NoArgsConstructor
@AllArgsConstructor
@Builder
public class ResultObj {
private String code;
private String message;
}10.測試
包括web請求和后臺任務
10.1 web請求
請求URL: http://127.0.0.1:8080/server/order/f1
入參：
{
"userName": "HangZhou0614",
"tradeName": "Vue進階教程"
}
返回值：
{
"code": "200",
"message": "成功"
}
10.2 后臺任務日志
日志輸出：
到此這篇關于Springboot詳解線程池與多線程及阻塞隊列的應用詳解的文章就介紹到這了,更多相關Springboot線程池內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
SpringBoot整合Hashids實現(xiàn)數據ID加密隱藏的全過程
這篇文章主要為大家詳細介紹了SpringBoot整合Hashids實現(xiàn)數據ID加密隱藏的全過程,文中的示例代碼講解詳細,對大家的學習或工作有一定的幫助,感興趣的小伙伴可以跟隨小編一起學習一下2024-01-01
SpringCloud?Gateway?DispatcherHandler調用方法詳細介紹
我們第一個關注的類就是DispatcherHandler，這個類提供的handle()方法，封裝了我們之后所有的handlerMappings，這個DispatcherHandler有點想SpringMVC的DispatchServlet，里面也是封裝了請求和對應的處理方法的關系2022-10-10
Spring事務有可能會提交，回滾、掛起、恢復，所以Spring事務提供了一種機制，可以讓程序員來監(jiān)聽當前Spring事務所處于的狀態(tài)，這篇文章主要介紹了Spring底層事務原理,需要的朋友可以參考下2022-12-12
Maven?項目用Assembly打包可執(zhí)行jar包的方法
這篇文章主要介紹了Maven?項目用Assembly打包可執(zhí)行jar包的方法,該方法只可打包非spring項目的可執(zhí)行jar包，需要的朋友可以參考下2023-03-03
項目中使用Springboot,在Controller中配置了@NotNull和@Valid,@Notnull不生效,@Valid生效,返回http?status為400,本文給大家介紹了Spring?controller校驗入參的方法,需要的朋友可以參考下2024-06-06
本文只是介紹五子棋AI的實現(xiàn)，最終的成品只是一個?AI?接口，并不包括?GUI，且不依賴?GUI，文中的示例代碼講解詳細，感興趣的可以嘗試一下2022-09-09
這篇文章主要介紹了spring boot 配置文件后綴的一個坑，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-09-09
這篇文章主要介紹了Jackson優(yōu)雅序列化Java枚舉類過程解析,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-09-09
java項目中常用指標UV?PV?QPS?TPS含義以及統(tǒng)計方法
文章介紹了現(xiàn)代Web應用中性能監(jiān)控和分析的重要性,涵蓋了UV、PV、QPS、TPS等關鍵指標的統(tǒng)計方法,并提供了示例代碼,同時,文章還討論了性能優(yōu)化和瓶頸分析的策略,以及使用Grafana等可視化工具進行監(jiān)控與告警的重要性2025-01-01
本文主要介紹了java.time包時間類的具體使用,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2024-08-08