快捷導(dǎo)航

java并發(fā)使用CountDownLatch在生產(chǎn)環(huán)境翻車剖析

更新時(shí)間：2022年08月10日 10:34:13 作者：叫我小郭_

這篇文章主要為大家介紹了java并發(fā)使用CountDownLatch在生產(chǎn)環(huán)境翻車的示例剖析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

前言

大家好，我是小郭，之前分享了CountDownLatch的使用，我們知道用來(lái)控制并發(fā)流程的同步工具，主要的作用是為了等待多個(gè)線程同時(shí)完成任務(wù)后，在進(jìn)行主線程任務(wù)。

萬(wàn)萬(wàn)沒(méi)想到，在生產(chǎn)環(huán)境中竟然翻車了，因?yàn)闆](méi)有考慮到一些場(chǎng)景，導(dǎo)致了CountDownLatch出現(xiàn)了問(wèn)題，接下來(lái)來(lái)分享一下由于CountDownLatch導(dǎo)致的問(wèn)題。

【線程】并發(fā)流程控制的同步工具-CountDownLatch

需求背景

先簡(jiǎn)單介紹下業(yè)務(wù)場(chǎng)景，針對(duì)用戶批量下載的文件進(jìn)行修改上傳

為了提高執(zhí)行的速度，所以在采用線程池去執(zhí)行下載-修改-上傳的操作，并在全部執(zhí)行完之后統(tǒng)一提交保存文件地址到數(shù)據(jù)庫(kù)，于是加入了CountDownLatch來(lái)進(jìn)行控制。

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

根據(jù)服務(wù)本身情況，自定義一個(gè)線程池

public static ExecutorService testExtcutor() {
        return new ThreadPoolExecutor(
                2,
                2,
                0L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1));
    }

模擬執(zhí)行

public static void main(String[] args) {
        // 下載文件總數(shù)
        List<Integer> resultList = new ArrayList<>(100);
        IntStream.range(0,100).forEach(resultList::add);
        // 下載文件分段
        List<List<Integer>> split = CollUtil.split(resultList, 10);
        ExecutorService executorService = BaseThreadPoolExector.testExtcutor();
        CountDownLatch countDownLatch = new CountDownLatch(100);
        for (List<Integer> list : split) {
            executorService.execute(() -> {
                list.forEach(i ->{
                    try {
                        // 模擬業(yè)務(wù)操作
                        Thread.sleep(500);
                        System.out.println("任務(wù)進(jìn)入");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        System.out.println(e.getMessage());
                    } finally {
                        System.out.println(countDownLatch.getCount());
                        countDownLatch.countDown();
                    }
                });
            });
        }
        try {
            countDownLatch.await();
            System.out.println("countDownLatch.await()");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

一開始我個(gè)人感覺(jué)沒(méi)有什么問(wèn)題，反正finally都能夠做減一的操作，到最后調(diào)用await方法，進(jìn)行主線程任務(wù)

Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@300ffa5d rejected from java.util.concurrent.ThreadPoolExecutor@1f17ae12[Running, pool size = 2, active threads = 2, queued tasks = 1, completed tasks = 0]
  at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
  at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
  at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
  at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:112)
  at Thread.executor.executorTestBlock.main(executorTestBlock.java:28)
任務(wù)進(jìn)入
countDownLatch.countDown
任務(wù)進(jìn)入
countDownLatch.countDown
任務(wù)進(jìn)入
countDownLatch.countDown
任務(wù)進(jìn)入
countDownLatch.countDown
任務(wù)進(jìn)入
countDownLatch.countDown
任務(wù)進(jìn)入
countDownLatch.countDown
任務(wù)進(jìn)入
countDownLatch.countDown
任務(wù)進(jìn)入
countDownLatch.countDown
任務(wù)進(jìn)入
countDownLatch.countDown
任務(wù)進(jìn)入
countDownLatch.countDown
任務(wù)進(jìn)入
countDownLatch.countDown
任務(wù)進(jìn)入
countDownLatch.countDown
任務(wù)進(jìn)入
countDownLatch.countDown
任務(wù)進(jìn)入
countDownLatch.countDown
任務(wù)進(jìn)入
countDownLatch.countDown

由于任務(wù)數(shù)量較多，阻塞隊(duì)列中已經(jīng)塞滿了，所以默認(rèn)的拒絕策略，當(dāng)隊(duì)列滿時(shí)，處理策略報(bào)錯(cuò)異常，

要注意這個(gè)異常是線程池，自己拋出的，不是我們循環(huán)里面打印出來(lái)的，

這也造成了，線上這個(gè)線程池被阻塞了，他永遠(yuǎn)也調(diào)用不到await方法，

利用jstack,我們就能夠看到有問(wèn)題

"pool-1-thread-2" #12 prio=5 os_prio=31 tid=0x00007ff6198b7000 nid=0xa903 waiting on condition [0x0000700001c64000]
   java.lang.Thread.State: WAITING (parking)
	at sun.misc.Unsafe.park(Native Method)
	- parking to wait for  <0x000000076b2283f8> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
	at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
	at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
	at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)
	at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
"pool-1-thread-1" #11 prio=5 os_prio=31 tid=0x00007ff6198b6800 nid=0x5903 waiting on condition [0x0000700001b61000]
   java.lang.Thread.State: WAITING (parking)
	at sun.misc.Unsafe.park(Native Method)
	- parking to wait for  <0x000000076b2283f8> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
	at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
	at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)
	at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:442)
	at java.util.concurrent.ThreadPoolExecutor.getTask(ThreadPoolExecutor.java:1074)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1134)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

解決方案

調(diào)大阻塞隊(duì)列，但是問(wèn)題來(lái)了，到底多少阻塞隊(duì)列才是大呢，如果太大了會(huì)不由又造成內(nèi)存溢出等其他的問(wèn)題

在第一個(gè)的基礎(chǔ)上，我們修改了拒絕策略,當(dāng)觸發(fā)拒絕策略的時(shí)候，用調(diào)用者所在的線程來(lái)執(zhí)行任務(wù)

 public static ThreadPoolExecutor queueExecutor(BlockingQueue<Runnable> workQueue){
        return new ThreadPoolExecutor(
                size,
                size,
                0L,
                TimeUnit.SECONDS,
                workQueue,
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

你可能又會(huì)想說(shuō)，會(huì)不會(huì)任務(wù)數(shù)量太多，導(dǎo)致調(diào)用者所在的線程執(zhí)行不過(guò)來(lái)，任務(wù)提交的性能急劇下降

那我們就應(yīng)該自定義拒絕策略，將這下排隊(duì)的消息記錄下來(lái)，采用補(bǔ)償機(jī)制的方式去執(zhí)行

同時(shí)也要注意上面的那個(gè)異常是線程池拋出來(lái)的，我們自己也需要將線程池進(jìn)行try catch，記錄問(wèn)題數(shù)據(jù)，并且在finally中執(zhí)行countDownLatch.countDown來(lái)避免，線程池的使用