快捷導(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)，直觀來(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)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: