快捷導(dǎo)航

java之使用多線程代替for循環(huán)(解決主線程提前結(jié)束問(wèn)題)

更新時(shí)間：2023年03月10日 08:55:24 作者：大聰明」

這篇文章主要介紹了java之使用多線程代替for循環(huán)(解決主線程提前結(jié)束問(wèn)題)，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教

CountDownLatch

CountDownLatch（也叫閉鎖）是一個(gè)同步協(xié)助類，允許一個(gè)或多個(gè)線程等待，直到其他線程完成操作集。
CountDownLatch 使用給定的計(jì)數(shù)值（count）初始化。await 方法會(huì)阻塞直到當(dāng)前的計(jì)數(shù)值（count）由于 countDown 方法的調(diào)用達(dá)到 0，count 為 0 之后所有等待的線程都會(huì)被釋放，并且隨后對(duì)await方法的調(diào)用都會(huì)立即返回。

構(gòu)造方法：

//參數(shù)count為計(jì)數(shù)值
public CountDownLatch(int count) {};

常用方法

// 調(diào)用 await() 方法的線程會(huì)被掛起，它會(huì)等待直到 count 值為 0 才繼續(xù)執(zhí)行
public void await() throws InterruptedException {};
 
// 和 await() 類似，若等待 timeout 時(shí)長(zhǎng)后，count 值還是沒(méi)有變?yōu)?0，不再等待，繼續(xù)執(zhí)行
public boolean await(long timeout, TimeUnit unit) throws InterruptedException {};
 
// 會(huì)將 count 減 1，直至為 0
public void countDown() {};

使用案例

首先是創(chuàng)建實(shí)例 CountDownLatch countDown = new CountDownLatch(2)；
需要同步的線程執(zhí)行完之后，計(jì)數(shù) -1， countDown.countDown()；
需要等待其他線程執(zhí)行完畢之后，再運(yùn)行的線程，調(diào)用 countDown.await()實(shí)現(xiàn)阻塞同步。
如下。

應(yīng)用場(chǎng)景

CountDownLatch 一般用作多線程倒計(jì)時(shí)計(jì)數(shù)器，強(qiáng)制它們等待其他一組（CountDownLatch的初始化決定）任務(wù)執(zhí)行完成。

CountDownLatch的兩種使用場(chǎng)景：

讓多個(gè)線程等待，模擬并發(fā)。
讓單個(gè)線程等待，多個(gè)線程（任務(wù)）完成后，進(jìn)行匯總合并。

場(chǎng)景1：模擬并發(fā)

import java.util.concurrent.CountDownLatch;
 
/**
 * 讓多個(gè)線程等待：模擬并發(fā)，讓并發(fā)線程一起執(zhí)行
 */
public class CountDownLatchTest {
    public static void main(String[] args) throws InterruptedException {
 
        CountDownLatch countDownLatch = new CountDownLatch(1);
        
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                try {
                    // 等待
                    countDownLatch.await();
                    String parter = "【" + Thread.currentThread().getName() + "】";
                    System.out.println(parter + "開(kāi)始執(zhí)行……");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
 
        Thread.sleep(2000);
       
        countDownLatch.countDown();
    }
}

場(chǎng)景2：多個(gè)線程完成后，進(jìn)行匯總合并

很多時(shí)候，我們的并發(fā)任務(wù)，存在前后依賴關(guān)系；比如數(shù)據(jù)詳情頁(yè)需要同時(shí)調(diào)用多個(gè)接口獲取數(shù)據(jù)，并發(fā)請(qǐng)求獲取到數(shù)據(jù)后、需要進(jìn)行結(jié)果合并；或者多個(gè)數(shù)據(jù)操作完成后，需要數(shù)據(jù) check；這其實(shí)都是：在多個(gè)線程(任務(wù))完成后，進(jìn)行匯總合并的場(chǎng)景。

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
 
/**
 * 讓單個(gè)線程等待：多個(gè)線程(任務(wù))完成后，進(jìn)行匯總合并
 */
public class CountDownLatchTest3 {
 
    //用于聚合所有的統(tǒng)計(jì)指標(biāo)
    private static Map map = new ConcurrentHashMap();
    //創(chuàng)建計(jì)數(shù)器，這里需要統(tǒng)計(jì)4個(gè)指標(biāo)
    private static CountDownLatch countDownLatch = new CountDownLatch(4);
 
    public static void main(String[] args) throws Exception {
 
        //記錄開(kāi)始時(shí)間
        long startTime = System.currentTimeMillis();
 
        Thread countUserThread = new Thread(() -> {
            try {
                System.out.println("正在統(tǒng)計(jì)新增用戶數(shù)量");
                Thread.sleep(3000);//任務(wù)執(zhí)行需要3秒
                map.put("userNumber", 100);//保存結(jié)果值
                System.out.println("統(tǒng)計(jì)新增用戶數(shù)量完畢");
                countDownLatch.countDown();//標(biāo)記已經(jīng)完成一個(gè)任務(wù)
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        Thread countOrderThread = new Thread(() -> {
            try {
                System.out.println("正在統(tǒng)計(jì)訂單數(shù)量");
                Thread.sleep(3000);//任務(wù)執(zhí)行需要3秒
                map.put("countOrder", 20);//保存結(jié)果值
                System.out.println("統(tǒng)計(jì)訂單數(shù)量完畢");
                countDownLatch.countDown();//標(biāo)記已經(jīng)完成一個(gè)任務(wù)
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
 
        Thread countGoodsThread = new Thread(() -> {
            try {
                System.out.println("正在商品銷量");
                Thread.sleep(3000);//任務(wù)執(zhí)行需要3秒
                map.put("countGoods", 300);//保存結(jié)果值
                System.out.println("統(tǒng)計(jì)商品銷量完畢");
                countDownLatch.countDown();//標(biāo)記已經(jīng)完成一個(gè)任務(wù)
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
 
        Thread countmoneyThread = new Thread(() -> {
            try {
                System.out.println("正在總銷售額");
                Thread.sleep(3000);//任務(wù)執(zhí)行需要3秒
                map.put("countMoney", 40000);//保存結(jié)果值
                System.out.println("統(tǒng)計(jì)銷售額完畢");
                countDownLatch.countDown();//標(biāo)記已經(jīng)完成一個(gè)任務(wù)
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        
        //啟動(dòng)子線程執(zhí)行任務(wù)
        countUserThread.start();
        countGoodsThread.start();
        countOrderThread.start();
        countmoneyThread.start();
 
        try {
            //主線程等待所有統(tǒng)計(jì)指標(biāo)執(zhí)行完畢
            countDownLatch.await();
            long endTime = System.currentTimeMillis();//記錄結(jié)束時(shí)間
            System.out.println("------統(tǒng)計(jì)指標(biāo)全部完成--------");
            System.out.println("統(tǒng)計(jì)結(jié)果為：" + map);
            System.out.println("任務(wù)總執(zhí)行時(shí)間為" + (endTime - startTime) + "ms");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
 
    }
}

接下來(lái)進(jìn)入正題

使用多線程代替for循環(huán)提高查詢效率，并且防止主線程提前結(jié)束導(dǎo)致其他線程數(shù)據(jù)錯(cuò)誤

直接上代碼：

@Override
    public AppResponse getLocations() throws InterruptedException {
        List<GetLocationVO> vos = new ArrayList<>();
        vos = projectDao.getLocationOne();    
//      原來(lái)的代碼
//        for (GetLocationVO vo : vos) {
//            List<LocationVO> children = projectDao.getLocationChildren(vo.getId());
//            vo.setChildren(children);
//        }
        //改造后的代碼
        Thread(vos,10);
        return AppResponse.success("查詢成功",vos);
    }
 
    //此處有加鎖
    public synchronized void Thread(List<GetLocationVO> list, int nThread) throws InterruptedException {
        if (CollectionUtils.isEmpty(list) || nThread <= 0 || CollectionUtils.isEmpty(list)) {
            return;
        }
        CountDownLatch latch = new CountDownLatch(list.size());//創(chuàng)建一個(gè)計(jì)數(shù)器（大小為當(dāng)前數(shù)組的大小，確保所有執(zhí)行完主線程才結(jié)束）
        ExecutorService pool = Executors.newFixedThreadPool(nThread);//創(chuàng)建一個(gè)固定的線程池
        for (GetLocationVO vo : list) {
            pool.execute(() -> {
                //處理的業(yè)務(wù)
                List<LocationVO> children = projectDao.getLocationChildren(vo.getId());
                vo.setChildren(children);
                latch.countDown();
            });
        }
        latch.await();
        pool.shutdown();
    }