快捷導(dǎo)航

詳解如何判斷Java線(xiàn)程池任務(wù)已執(zhí)行完

更新時(shí)間：2023年08月04日 11:11:47 作者：Java中文社群

線(xiàn)程池的使用并不復(fù)雜，麻煩的是如何判斷線(xiàn)程池中的任務(wù)已經(jīng)全部執(zhí)行完了，所以接下來(lái)，我們就來(lái)看看如何判斷線(xiàn)程中的任務(wù)是否已經(jīng)全部執(zhí)行完吧

例如某個(gè)復(fù)雜的查詢(xún)，無(wú)法使用一個(gè)查詢(xún)語(yǔ)句來(lái)完成此功能，此時(shí)我們就需要執(zhí)行多個(gè)查詢(xún)語(yǔ)句，然后再將各自查詢(xún)的結(jié)果，組裝之后返回給前端了，那么這種場(chǎng)景下，我們就必須使用線(xiàn)程池來(lái)進(jìn)行并發(fā)查詢(xún)了。

PS：磊哥做的最復(fù)雜的查詢(xún)，總共關(guān)聯(lián)了 21 張表，在和產(chǎn)品及需求方的溝通多次溝通下，才將查詢(xún)的業(yè)務(wù)從 21 張表，降到了至少要查詢(xún) 12 張表（非常難搞），那么這種場(chǎng)景下是無(wú)法使用一個(gè)查詢(xún)語(yǔ)句來(lái)實(shí)現(xiàn)的，那么并發(fā)查詢(xún)是必須要給安排上的。

1.需求分析

線(xiàn)程池的使用并不復(fù)雜，麻煩的是如何判斷線(xiàn)程池中的任務(wù)已經(jīng)全部執(zhí)行完了？因?yàn)槲覀円人腥蝿?wù)都執(zhí)行完之后，才能進(jìn)行數(shù)據(jù)的組裝和返回，所以接下來(lái)，我們就來(lái)看如何判斷線(xiàn)程中的任務(wù)是否已經(jīng)全部執(zhí)行完？

2.實(shí)現(xiàn)概述

判斷線(xiàn)程池中的任務(wù)是否執(zhí)行完的方法有很多，比如以下幾個(gè)：

使用 getCompletedTaskCount() 統(tǒng)計(jì)已經(jīng)執(zhí)行完的任務(wù)，和 getTaskCount() 線(xiàn)程池的總?cè)蝿?wù)進(jìn)行對(duì)比，如果相等則說(shuō)明線(xiàn)程池的任務(wù)執(zhí)行完了，否則既未執(zhí)行完。
使用 FutureTask 等待所有任務(wù)執(zhí)行完，線(xiàn)程池的任務(wù)就執(zhí)行完了。
使用 CountDownLatch 或 CyclicBarrier 等待所有線(xiàn)程都執(zhí)行完之后，再執(zhí)行后續(xù)流程。

具體實(shí)現(xiàn)代碼如下。

3.具體實(shí)現(xiàn)

3.1 統(tǒng)計(jì)完成任務(wù)數(shù)

通過(guò)判斷線(xiàn)程池中的計(jì)劃執(zhí)行任務(wù)數(shù)和已完成任務(wù)數(shù)，來(lái)判斷線(xiàn)程池是否已經(jīng)全部執(zhí)行完，如果計(jì)劃執(zhí)行任務(wù)數(shù)=已完成任務(wù)數(shù)，那么線(xiàn)程池的任務(wù)就全部執(zhí)行完了，否則就未執(zhí)行完。示例代碼如下：

private?static?void?isCompletedByTaskCount(ThreadPoolExecutor?threadPool)?{
????while?(threadPool.getTaskCount()?!=?threadPool.getCompletedTaskCount())?{
????}
}

以上程序執(zhí)行結(jié)果如下：

方法說(shuō)明

getTaskCount()：返回計(jì)劃執(zhí)行的任務(wù)總數(shù)。由于任務(wù)和線(xiàn)程的狀態(tài)可能在計(jì)算過(guò)程中動(dòng)態(tài)變化，因此返回的值只是一個(gè)近似值。
getCompletedTaskCount()：返回完成執(zhí)行任務(wù)的總數(shù)。因?yàn)槿蝿?wù)和線(xiàn)程的狀態(tài)可能在計(jì)算過(guò)程中動(dòng)態(tài)地改變，所以返回的值只是一個(gè)近似值，但是在連續(xù)的調(diào)用中并不會(huì)減少。

缺點(diǎn)分析

此判斷方法的缺點(diǎn)是 getTaskCount() 和 getCompletedTaskCount() 返回的是一個(gè)近似值，因?yàn)榫€(xiàn)程池中的任務(wù)和線(xiàn)程的狀態(tài)可能在計(jì)算過(guò)程中動(dòng)態(tài)變化，所以它們兩個(gè)返回的都是一個(gè)近似值。

3.2 FutureTask

FutrueTask 的優(yōu)勢(shì)是任務(wù)判斷精準(zhǔn)，調(diào)用每個(gè) FutrueTask 的 get 方法就是等待該任務(wù)執(zhí)行完，如下代碼所示：

import?java.util.concurrent.ExecutionException;
import?java.util.concurrent.ExecutorService;
import?java.util.concurrent.Executors;
import?java.util.concurrent.FutureTask;
/**
?*?使用?FutrueTask?等待線(xiàn)程池執(zhí)行完全部任務(wù)
?*/
public?class?FutureTaskDemo?{
????public?static?void?main(String[]?args)?throws?ExecutionException,?InterruptedException?{
????????//?創(chuàng)建一個(gè)固定大小的線(xiàn)程池
????????ExecutorService?executor?=?Executors.newFixedThreadPool(3);
????????//?創(chuàng)建任務(wù)
????????FutureTask<Integer>?task1?=?new?FutureTask<>(()?->?{
????????????System.out.println("Task?1?start");
????????????Thread.sleep(2000);
????????????System.out.println("Task?1?end");
????????????return?1;
????????});
????????FutureTask<Integer>?task2?=?new?FutureTask<>(()?->?{
????????????System.out.println("Task?2?start");
????????????Thread.sleep(3000);
????????????System.out.println("Task?2?end");
????????????return?2;
????????});
????????FutureTask<Integer>?task3?=?new?FutureTask<>(()?->?{
????????????System.out.println("Task?3?start");
????????????Thread.sleep(1500);
????????????System.out.println("Task?3?end");
????????????return?3;
????????});
????????//?提交三個(gè)任務(wù)給線(xiàn)程池
????????executor.submit(task1);
????????executor.submit(task2);
????????executor.submit(task3);
????????//?等待所有任務(wù)執(zhí)行完畢并獲取結(jié)果
????????int?result1?=?task1.get();
????????int?result2?=?task2.get();
????????int?result3?=?task3.get();
????????System.out.println("Do?main?thread.");
????}
}

以上程序的執(zhí)行結(jié)果如下：

3.3 CountDownLatch和CyclicBarrier

CountDownLatch 和 CyclicBarrier 類(lèi)似，都是等待所有任務(wù)到達(dá)某個(gè)點(diǎn)之后，再進(jìn)行后續(xù)的操作，如下圖所示：

CountDownLatch 使用的示例代碼如下：

public?static?void?main(String[]?args)?throws?InterruptedException?{
????//?創(chuàng)建線(xiàn)程池
????ThreadPoolExecutor?threadPool?=?new?ThreadPoolExecutor(10,?20,
?????0,?TimeUnit.SECONDS,?new?LinkedBlockingDeque<>(1024));
????final?int?taskCount?=?5;????//?任務(wù)總數(shù)
????//?單次計(jì)數(shù)器
????CountDownLatch?countDownLatch?=?new?CountDownLatch(taskCount);?//?①
????//?添加任務(wù)
????for?(int?i?=?0;?i?<?taskCount;?i++)?{
????????final?int?finalI?=?i;
????????threadPool.submit(new?Runnable()?{
????????????@Override
????????????public?void?run()?{
????????????????try?{
????????????????????//?隨機(jī)休眠?0-4s
????????????????????int?sleepTime?=?new?Random().nextInt(5);
????????????????????TimeUnit.SECONDS.sleep(sleepTime);
????????????????}?catch?(InterruptedException?e)?{
????????????????????e.printStackTrace();
????????????????}
????????????????System.out.println(String.format("任務(wù)%d執(zhí)行完成",?finalI));
????????????????//?線(xiàn)程執(zhí)行完，計(jì)數(shù)器?-1
????????????????countDownLatch.countDown();??//?②
????????????}
????????});
????}
????//?阻塞等待線(xiàn)程池任務(wù)執(zhí)行完
????countDownLatch.await();??//?③
????//?線(xiàn)程池執(zhí)行完
????System.out.println();
????System.out.println("線(xiàn)程池任務(wù)執(zhí)行完成！");
}

代碼說(shuō)明：以上代碼中標(biāo)識(shí)為 ①、②、③ 的代碼行是核心實(shí)現(xiàn)代碼，其中：① 是聲明一個(gè)包含了 5 個(gè)任務(wù)的計(jì)數(shù)器；② 是每個(gè)任務(wù)執(zhí)行完之后計(jì)數(shù)器 -1；③ 是阻塞等待計(jì)數(shù)器 CountDownLatch 減為 0，表示任務(wù)都執(zhí)行完了，可以執(zhí)行 await 方法后面的業(yè)務(wù)代碼了。

以上程序的執(zhí)行結(jié)果如下：

缺點(diǎn)分析

CountDownLatch 缺點(diǎn)是計(jì)數(shù)器只能使用一次，CountDownLatch 創(chuàng)建之后不能被重復(fù)使用。CyclicBarrier 和 CountDownLatch 類(lèi)似，它可以理解為一個(gè)可以重復(fù)使用的循環(huán)計(jì)數(shù)器，CyclicBarrier 可以調(diào)用 reset 方法將自己重置到初始狀態(tài)，CyclicBarrier 具體實(shí)現(xiàn)代碼如下：

public?static?void?main(String[]?args)?throws?InterruptedException?{
????//?創(chuàng)建線(xiàn)程池
????ThreadPoolExecutor?threadPool?=?new?ThreadPoolExecutor(10,?20,
?????0,?TimeUnit.SECONDS,?new?LinkedBlockingDeque<>(1024));
????final?int?taskCount?=?5;????//?任務(wù)總數(shù)
????//?循環(huán)計(jì)數(shù)器?①
????CyclicBarrier?cyclicBarrier?=?new?CyclicBarrier(taskCount,?new?Runnable()?{
????????@Override
????????public?void?run()?{
????????????//?線(xiàn)程池執(zhí)行完
????????????System.out.println();
????????????System.out.println("線(xiàn)程池所有任務(wù)已執(zhí)行完！");
????????}
????});
????//?添加任務(wù)
????for?(int?i?=?0;?i?<?taskCount;?i++)?{
????????final?int?finalI?=?i;
????????threadPool.submit(new?Runnable()?{
????????????@Override
????????????public?void?run()?{
????????????????try?{
????????????????????//?隨機(jī)休眠?0-4s
????????????????????int?sleepTime?=?new?Random().nextInt(5);
????????????????????TimeUnit.SECONDS.sleep(sleepTime);
????????????????????System.out.println(String.format("任務(wù)%d執(zhí)行完成",?finalI));
????????????????????//?線(xiàn)程執(zhí)行完
????????????????????cyclicBarrier.await();?//?②
????????????????}?catch?(InterruptedException?e)?{
????????????????????e.printStackTrace();
????????????????}?catch?(BrokenBarrierException?e)?{
????????????????????e.printStackTrace();
????????????????}
????????????}
????????});
????}
}

以上程序的執(zhí)行結(jié)果如下：

方法說(shuō)明

CyclicBarrier 有 3 個(gè)重要的方法：

1.構(gòu)造方法：構(gòu)造方法可以傳遞兩個(gè)參數(shù)，參數(shù) 1 是計(jì)數(shù)器的數(shù)量 parties，參數(shù) 2 是計(jì)數(shù)器為 0 時(shí)，也就是任務(wù)都執(zhí)行完之后可以執(zhí)行的事件（方法）。

2.await 方法：在 CyclicBarrier 上進(jìn)行阻塞等待，當(dāng)調(diào)用此方法時(shí) CyclicBarrier 的內(nèi)部計(jì)數(shù)器會(huì) -1，直到發(fā)生以下情形之一：

在 CyclicBarrier 上等待的線(xiàn)程數(shù)量達(dá)到 parties，也就是計(jì)數(shù)器的聲明數(shù)量時(shí)，則所有線(xiàn)程被釋放，繼續(xù)執(zhí)行。
當(dāng)前線(xiàn)程被中斷，則拋出 InterruptedException 異常，并停止等待，繼續(xù)執(zhí)行。
其他等待的線(xiàn)程被中斷，則當(dāng)前線(xiàn)程拋出 BrokenBarrierException 異常，并停止等待，繼續(xù)執(zhí)行。
其他等待的線(xiàn)程超時(shí)，則當(dāng)前線(xiàn)程拋出 BrokenBarrierException 異常，并停止等待，繼續(xù)執(zhí)行。
其他線(xiàn)程調(diào)用 CyclicBarrier.reset() 方法，則當(dāng)前線(xiàn)程拋出 BrokenBarrierException 異常，并停止等待，繼續(xù)執(zhí)行。

3.reset 方法：使得CyclicBarrier回歸初始狀態(tài)，直觀(guān)來(lái)看它做了兩件事：

如果有正在等待的線(xiàn)程，則會(huì)拋出 BrokenBarrierException 異常，且這些線(xiàn)程停止等待，繼續(xù)執(zhí)行。
將是否破損標(biāo)志位 broken 置為 false。

優(yōu)缺點(diǎn)分析

CyclicBarrier 從設(shè)計(jì)的復(fù)雜度到使用的復(fù)雜度都高于 CountDownLatch，相比于 CountDownLatch 來(lái)說(shuō)它的優(yōu)點(diǎn)是可以重復(fù)使用（只需調(diào)用 reset 就能恢復(fù)到初始狀態(tài)），缺點(diǎn)是使用難度較高。

小結(jié)

在實(shí)現(xiàn)判斷線(xiàn)程池任務(wù)是否執(zhí)行完成的方案中，通過(guò)統(tǒng)計(jì)線(xiàn)程池執(zhí)行完任務(wù)的方式（實(shí)現(xiàn)方法 1），以及實(shí)現(xiàn)方法 3（CountDownLatch 或 CyclicBarrier）等統(tǒng)計(jì)，都是“不記名”的，只關(guān)注數(shù)量，不關(guān)注（具體）對(duì)象，所以這些方式都有可能受到外界代碼的影響，因此使用 FutureTask 等待具體任務(wù)執(zhí)行完的方式是最推薦的判斷方法。

到此這篇關(guān)于詳解如何判斷Java線(xiàn)程池任務(wù)已執(zhí)行完的文章就介紹到這了,更多相關(guān)Java線(xiàn)程池任務(wù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: