新來個阿里 P7，僅花 2 小時，做出一個多執行緒永動任務，看完直接跪了
大家好，我是樓仔！
今天教大家做一個 Java 的多執行緒永動任務，這個示例的原型是公司自研的多執行緒非同步任務專案，我把裡面涉及到多執行緒的程式碼抽離出來，然後進行一定的改造。
裡面涉及的知識點非常多，特別適合有一定工作經驗的同學學習，或者可以直接拿到專案中使用。
文章結構非常簡單：
1. 功能說明
做這個多執行緒非同步任務，主要是因為我們有很多永動的非同步任務，什麼是永動呢？就是任務跑起來後，需要一直跑下去。
比如訊息 Push 任務，因為一直有訊息過來，所以需要一直去消費 DB 中的未推送訊息，就需要整一個 Push 的永動非同步任務。
我們的需求其實不難，簡單總結一下：
- 能同時執行多個永動的非同步任務；
- 每個非同步任務，支援開多個執行緒去消費這個任務的資料；
- 支援永動非同步任務的優雅關閉，即關閉後，需要把所有的資料消費完畢後，再關閉。
完成上面的需求，需要注意幾個點：
- 每個永動任務，可以開一個執行緒去執行；
- 每個子任務，因為需要支援併發，需要用執行緒池控制；
- 永動任務的關閉，需要通知子任務的併發執行緒，並支援永動任務和併發子任務的優雅關閉。
2. 多執行緒任務示例
2.1 執行緒池
對於子任務，需要支援併發，如果每個併發都開一個執行緒，用完就關閉，對資源消耗太大，所以引入執行緒池：
public class TaskProcessUtil {
private static Map<String, ExecutorService> executors = new ConcurrentHashMap<>();
private static ExecutorService init(String poolName, int poolSize) {
return new ThreadPoolExecutor(poolSize, poolSize,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>(),
new ThreadFactoryBuilder().setNameFormat("Pool-" + poolName).setDaemon(false).build(),
new ThreadPoolExecutor.CallerRunsPolicy());
}
public static ExecutorService getOrInitExecutors(String poolName,int poolSize) {
ExecutorService executorService = executors.get(poolName);
if (null == executorService) {
synchronized (TaskProcessUtil.class) {
executorService = executors.get(poolName);
if (null == executorService) {
executorService = init(poolName, poolSize);
executors.put(poolName, executorService);
}
}
}
return executorService;
}
public static void releaseExecutors(String poolName) {
ExecutorService executorService = executors.remove(poolName);
if (executorService != null) {
executorService.shutdown();
}
}
}
這是一個執行緒池的工具類，這裡初始化執行緒池和回收執行緒資源很簡單，我們主要討論獲取執行緒池。
獲取執行緒池可能會存在併發情況，所以需要加一個 synchronized 鎖，然後鎖住後，需要對 executorService 進行二次判空校驗。
2.2 單個任務
為了更好講解單個任務的實現方式，我們的任務主要就是把 Cat 的資料打印出來，Cat 定義如下：
@Data
@Service
public class Cat {
private String catName;
public Cat setCatName(String name) {
this.catName = name;
return this;
}
}
單個任務主要包括以下功能：
獲取永動任務資料：這裡一般都是掃描 DB，我直接就簡單用 queryData() 代替。
多執行緒執行任務：需要把資料拆分成 4 份，然後分別由多執行緒併發執行，這裡可以通過執行緒池支援；
永動任務優雅停機：當外面通知任務需要停機，需要執行完剩餘任務資料，並回收執行緒資源，退出任務；
永動執行：如果未收到停機命令，任務需要一直執行下去。
直接看程式碼：
public class ChildTask {
private final int POOL_SIZE = 3;
private final int SPLIT_SIZE = 4;
private String taskName;
protected volatile boolean terminal = false;
public ChildTask(String taskName) {
this.taskName = taskName;
}
public void doExecute() {
int i = 0;
while(true) {
System.out.println(taskName + ":Cycle-" + i + "-Begin");
List<Cat> datas = queryData();
taskExecute(datas);
System.out.println(taskName + ":Cycle-" + i + "-End");
if (terminal) {
break;
}
i++;
}
TaskProcessUtil.releaseExecutors(taskName);
}
public void terminal() {
terminal = true;
System.out.println(taskName + " shut down");
}
private void doProcessData(List<Cat> datas, CountDownLatch latch) {
try {
for (Cat cat : datas) {
System.out.println(taskName + ":" + cat.toString() + ",ThreadName:" + Thread.currentThread().getName());
Thread.sleep(1000L);
}
} catch (Exception e) {
System.out.println(e.getStackTrace());
} finally {
if (latch != null) {
latch.countDown();
}
}
}
private void taskExecute(List<Cat> sourceDatas) {
if (CollectionUtils.isEmpty(sourceDatas)) {
return;
}
List<List<Cat>> splitDatas = Lists.partition(sourceDatas, SPLIT_SIZE);
final CountDownLatch latch = new CountDownLatch(splitDatas.size());
for (final List<Cat> datas : splitDatas) {
ExecutorService executorService = TaskProcessUtil.getOrInitExecutors(taskName, POOL_SIZE);
executorService.submit(new Runnable() {
@Override
public void run() {
doProcessData(datas, latch);
}
});
}
try {
latch.await();
} catch (Exception e) {
System.out.println(e.getStackTrace());
}
}
private List<Cat> queryData() {
List<Cat> datas = new ArrayList<>();
for (int i = 0; i < 5; i ++) {
datas.add(new Cat().setCatName("羅小黑" + i));
}
return datas;
}
}
簡單解釋一下：
- queryData：用於獲取資料，實際應用中其實是需要把 queryData 定為抽象方法，然後由各個任務實現自己的方法。
- doProcessData：資料處理邏輯，實際應用中其實是需要把 doProcessData 定為抽象方法，然後由各個任務實現自己的方法。
- taskExecute：將資料拆分成 4 份，獲取該任務的執行緒池，並交給執行緒池併發執行，然後通過 latch.await() 阻塞。當這 4 份資料都執行成功後，阻塞結束，該方法才返回。
- terminal：僅用於接受停機命令，這裡該變數定義為 volatile，所以多執行緒記憶體可見；
doExecute：程式執行入口，封裝了每個任務執行的流程，當 terminal=true 時，先執行完任務資料，然後回收執行緒池，最後退出。
2.3 任務入口
直接上程式碼：
public class LoopTask {
private List<ChildTask> childTasks;
public void initLoopTask() {
childTasks = new ArrayList();
childTasks.add(new ChildTask("childTask1"));
childTasks.add(new ChildTask("childTask2"));
for (final ChildTask childTask : childTasks) {
new Thread(new Runnable() {
@Override
public void run() {
childTask.doExecute();
}
}).start();
}
}
public void shutdownLoopTask() {
if (!CollectionUtils.isEmpty(childTasks)) {
for (ChildTask childTask : childTasks) {
childTask.terminal();
}
}
}
public static void main(String args[]) throws Exception{
LoopTask loopTask = new LoopTask();
loopTask.initLoopTask();
Thread.sleep(5000L);
loopTask.shutdownLoopTask();
}
}
每個任務都開一個單獨的 Thread，這裡我初始化了 2 個永動任務，分別為 childTask1 和 childTask2，然後分別執行，後面 Sleep 了 5 秒後，再關閉任務，我們可以看看是否可以按照我們的預期優雅退出。
2.4 結果分析
執行結果如下：
childTask1:Cycle-0-Begin childTask2:Cycle-0-Begin childTask1:Cat(catName=羅小黑0),ThreadName:Pool-childTask1 childTask1:Cat(catName=羅小黑4),ThreadName:Pool-childTask1 childTask2:Cat(catName=羅小黑4),ThreadName:Pool-childTask2 childTask2:Cat(catName=羅小黑0),ThreadName:Pool-childTask2 childTask1:Cat(catName=羅小黑1),ThreadName:Pool-childTask1 childTask2:Cat(catName=羅小黑1),ThreadName:Pool-childTask2 childTask2:Cat(catName=羅小黑2),ThreadName:Pool-childTask2 childTask1:Cat(catName=羅小黑2),ThreadName:Pool-childTask1 childTask2:Cat(catName=羅小黑3),ThreadName:Pool-childTask2 childTask1:Cat(catName=羅小黑3),ThreadName:Pool-childTask1 childTask2:Cycle-0-End childTask2:Cycle-1-Begin childTask1:Cycle-0-End childTask1:Cycle-1-Begin childTask2:Cat(catName=羅小黑0),ThreadName:Pool-childTask2 childTask2:Cat(catName=羅小黑4),ThreadName:Pool-childTask2 childTask1:Cat(catName=羅小黑4),ThreadName:Pool-childTask1 childTask1:Cat(catName=羅小黑0),ThreadName:Pool-childTask1 childTask1 shut down childTask2 shut down childTask2:Cat(catName=羅小黑1),ThreadName:Pool-childTask2 childTask1:Cat(catName=羅小黑1),ThreadName:Pool-childTask1 childTask1:Cat(catName=羅小黑2),ThreadName:Pool-childTask1 childTask2:Cat(catName=羅小黑2),ThreadName:Pool-childTask2 childTask1:Cat(catName=羅小黑3),ThreadName:Pool-childTask1 childTask2:Cat(catName=羅小黑3),ThreadName:Pool-childTask2 childTask1:Cycle-1-End childTask2:Cycle-1-End
輸出資料：
- “Pool-childTask” 是執行緒池名稱；
- “childTask” 是任務名稱；
- “Cat(catName=羅小黑)” 是執行的結果；
- “childTask shut down” 是關閉標記；
- “childTask:Cycle-X-Begin” 和“childTask:Cycle-X-End” 是每一輪迴圈的開始和結束標記。
我們分析一下執行結果：
- childTask1 和 childTask2 分別執行，在第一輪迴圈中都正常輸出了 5 條羅小黑資料；
- 第二輪執行過程中，我啟動了關閉指令，這次第二輪執行沒有直接停止，而是先執行完任務中的資料，再執行退出，所以完全符合我們的優雅退出結論。
2.5 原始碼地址
GitHub 地址：
http://github.com/lml200701158/java-study/tree/master/src/main/java/com/java/parallel/pool/ofc
3. 寫在最後
對於這個經典的執行緒池使用示例，原專案是我好友一灰寫的，技術水平對標阿里 P7，實現得也非常優雅，涉及的知識點非常多，非常值得大家學習。
「其他文章」
- Spring中實現非同步呼叫的方式有哪些？
- 帶引數的全型別 Python 裝飾器
- 整理了幾個Python正則表示式，拿走就能用！
- SOLID：開閉原則Go程式碼實戰
- React中如何引入CSS呢
- 一個新視角：前端框架們都卷錯方向了？
- 編碼中的Adapter，不僅是一種設計模式，更是一種架構理念與解決方案
- 手寫程式語言-遞迴函式是如何實現的？
- 一文搞懂模糊匹配：定義、過程與技術
- 新來個阿里 P7，僅花 2 小時，做出一個多執行緒永動任務，看完直接跪了
- Puzzlescript，一種開發H5益智遊戲的引擎
- @Autowired和@Resource到底什麼區別，你明白了嗎？
- CSS transition 小技巧！如何保留 hover 的狀態？
- React如此受歡迎離不開這4個主要原則
- LeCun再炮轟Marcus： 他是心理學家，不是搞AI的
- Java保證執行緒安全的方式有哪些？
- 19個殺手級 JavaScript 單行程式碼，讓你看起來像專業人士
- Python 的"self"引數是什麼？
- 別整一坨 CSS 程式碼了，試試這幾個實用函式
- 再有人問你什麼是MVCC，就把這篇文章發給他！