java中的并發(fā)工具類

一：等待多線程完成的CountDownLatch

CountDownLatch允許一個或多個線程等待其他線程完成操作。

package com.fuzhulei;import java.util.concurrent.*;​/** * 減法計數(shù)器，主要是countDown（計數(shù)器1） 和 await（阻塞）方法，只有當計數(shù)器減為0的時候，當前線程才可以往下繼續(xù)執(zhí)行。 * 主要用于允許一個或多個線程等待其他線程完成操作 * @author Huxudong * @createTime 2020-04-05 00:04:36 **/public class CountDownDemo { public static void main(String[] args) throws InterruptedException { /** 使用其構(gòu)造函數(shù)，創(chuàng)建一個數(shù)值為6的計數(shù)器 */ CountDownLatch countDownLatch = new CountDownLatch(6); /** 自定義線程池使用 */ ExecutorService pool = new ThreadPoolExecutor(6, // 核心線程池大小9, // 最大線程池的大小（根據(jù)是IO密集型，還是CPU密集型來確定大小）3L, // 超時等待時間TimeUnit.SECONDS, // 時間的單位new LinkedBlockingQueue<>(5), // 阻塞隊列是哪一種Executors.defaultThreadFactory(), // 默認線程創(chuàng)建工廠new ThreadPoolExecutor.AbortPolicy() // 四大拒絕策略，選擇一種 ); try{ for (int i = 0; i < 6; i++) {/** 這個線程的提交，沒有返回值的任務(wù) */pool.execute(()->{ countDownLatch.countDown(); System.out.println(Thread.currentThread().getName()+'執(zhí)行一次減法');});​ } } catch(Exception e) { e.printStackTrace(); } finally { /** 關(guān)閉線程池 */ pool.shutdown(); }​ countDownLatch.await(); System.out.println('執(zhí)行完成了');​ }}

正確執(zhí)行結(jié)果：

但是如果我們設(shè)置計數(shù)器的容量大于6的話（相對于我的程序而言），就會被阻塞在那里

會發(fā)現(xiàn)執(zhí)行完成了 沒有被打印出來，而且程序一直沒有停止，這個時候就是因為計數(shù)器沒有歸0，所以當前線程被阻塞，不能向下面繼續(xù)進行。

二：同步屏障CyclicBarrier

CyclicBarrier的翻譯大致就是可循環(huán)的屏障。它主要的作用就是讓一組線程到達一個屏障（也可以叫做同步點）時被阻塞，直到最后一份線程到達屏障時，屏障才會開門，所有被屏障攔截的線程才會繼續(xù)運行。

package com.fuzhulei;​import java.util.concurrent.BrokenBarrierException;import java.util.concurrent.CyclicBarrier;​/** * CyclicBarrier是一個加法計數(shù)器，即同步屏障，可循環(huán)的屏障，讓一組線程到達一個屏障（也可以叫做同步點）時被阻塞，直到最后一個線程到達屏障，達到了一開始初始化的屏障的數(shù)值， * 屏障才可以打開門，所有被攔截的線程才可以繼續(xù)工作，主要是通過調(diào)用await方法來實現(xiàn)的 * @author Huxudong * @createTime 2020-04-04 22:53:50 **/public class CyclicBarrierDemo { public static void main(String[] args) { CyclicBarrier cyclicBarrier = new CyclicBarrier(3); new Thread(()->{​ try {cyclicBarrier.await(); } catch (InterruptedException e) {e.printStackTrace(); } catch (BrokenBarrierException e) {e.printStackTrace(); } System.out.println('線程A已經(jīng)到達屏障'); },'A').start();​ new Thread(()->{ try {cyclicBarrier.await(); } catch (InterruptedException e) {e.printStackTrace(); } catch (BrokenBarrierException e) {e.printStackTrace(); } System.out.println('線程B已經(jīng)到達屏障'); },'B').start();​ new Thread(()->{ try {cyclicBarrier.await(); } catch (InterruptedException e) {e.printStackTrace(); } catch (BrokenBarrierException e) {e.printStackTrace(); } System.out.println('線程C已經(jīng)到達屏障'); },'C').start();​​ }}

執(zhí)行的結(jié)果如下：

但是如果把定義的容量大于3（相對于我的程序而言），就會發(fā)現(xiàn)什么都不會輸出了，看截圖

并且程序一直還沒有停止，這就是屏障起到了作用，因為屏障要求至少需要4個（假設(shè)），但是此時只有三個線程到達，所以不滿足，屏障就一直阻攔不放路，那么所有的線程也就被阻塞不能向下面繼續(xù)運行，除非知道第四個過來，滿足條件才會運行。

三：控制并發(fā)線程數(shù)的Semaphore

用來控制同時訪問特定資源的線程數(shù)量，通過協(xié)調(diào)各個線程，以保證合理的使用公用的資源。

package com.fuzhulei;​import java.util.concurrent.*;​/** * 用來控制同時訪問特定資源的線程數(shù)量，通過協(xié)調(diào)各個線程，以保證合理的使用公用的資源 * @author Huxudong * @createTime 2020-04-04 23:45:29 **/public class SemaphoreDemo { public static void main(String[] args) { Semaphore semaphore = new Semaphore(5); ExecutorService pool = new ThreadPoolExecutor(10,20,3L,TimeUnit.SECONDS,new LinkedBlockingQueue<>(20),Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());​ try{ for (int i = 0; i < 60; i++) {pool.execute(() ->{ try { semaphore.acquire(); System.out.println(Thread.currentThread().getName()+'限流成功'); semaphore.release(); } catch (InterruptedException e) { e.printStackTrace(); }}); } } catch(Exception e) { e.printStackTrace(); } finally { pool.shutdown(); } }}

執(zhí)行的結(jié)果如下：

例如：數(shù)據(jù)庫資源，假如需要讀取幾十萬個數(shù)據(jù)的文件，因為都是IO密集型任務(wù)，所以開了2倍的處理器+1個線程數(shù)（IO密集型，所以線程可以多一些，讓cpu忙起來，因為IO操作的時候，很少操作Cpu）

但是如果讀到內(nèi)存后，還需要存儲到數(shù)據(jù)庫中，但是數(shù)據(jù)庫連接我們設(shè)置的加入就10個，所以我們必須控制只有10個線程可以同時訪問數(shù)據(jù)庫連接保存數(shù)據(jù)，否則會報錯無法連接數(shù)據(jù)庫異常。

以上就是本文的全部內(nèi)容，希望對大家的學習有所幫助，也希望大家多多支持好吧啦網(wǎng)。