在JDK的并發(fā)包里面提供了幾個非常有用的并發(fā)工具,CountDownLatch、CyclicBarrier、Semaphore工具類提供了一種并發(fā)控制流程的手段,Exchanger工具類則提供了在線程間交換數(shù)據(jù)的一種手段。
1、等待多線程完成的CountDownLatch
CountDownLatch是一個同步輔助類,在完成一組正在其他線程中執(zhí)行的操作之前,它運(yùn)行一個或者多個線程一直處于等待狀態(tài)。
CountDownLatch中有兩個關(guān)鍵的方法
public void countDown() {}
public boolean await(long timeout, TimeUnit unit){}
CountDownLatch是一個計數(shù)器,在它的構(gòu)造方法中需要指定一個值,用來設(shè)定計數(shù)的次數(shù)。
每調(diào)用一次countDown()方法,數(shù)值便會減一,CountDownLatch會一直阻塞著調(diào)用await()方法的線程
直到計數(shù)器的值變?yōu)?。
設(shè)想有這樣一個功能需要Thread1、Thread2、Thread3、Thread4四條線程分別統(tǒng)計C、D、E、F
四個盤的大小,所有線程都統(tǒng)計完畢交給主線程去做匯總,利用CountDownLatch來完成就非常輕松。
在實(shí)際的項(xiàng)目開發(fā)中,有類似的應(yīng)用場景用CountDownLatch來實(shí)現(xiàn)也非常簡單實(shí)用。
package com.dreyer.javadoc.thread;
import java.util.Date;
import java.util.Random;
import java.util.concurrent.*;
/**
* @description CountDownLatch
* @author: 會跳舞的機(jī)器人
* @date: 16/5/14 下午11:41
*/
public class CountDownLatchDemo {
/**
*
*/
private static CountDownLatch countDownLatch = new CountDownLatch(4);
/**
* 線程池
*/
private static ExecutorService executor = Executors.newFixedThreadPool(4);
/**
* 開啟的線程數(shù)
*/
private static int THREAD_COUNT = 4;
public static void main(String[] args) throws InterruptedException {
for (int i = 0; i < THREAD_COUNT; i++) {
executor.execute(new Runnable() {
public void run() {
try {
// 模擬業(yè)務(wù)邏輯的耗時
int timer = new Random().nextInt(5);
TimeUnit.SECONDS.sleep(timer);
System.out.printf("%s時完成磁盤的統(tǒng)計任務(wù),耗費(fèi)%d秒.\n", new Date().toString(), timer);
// 業(yè)務(wù)處理完成之后,計數(shù)器減一
countDownLatch.countDown();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
// 主線程一直被阻塞,直到countDownLatch的值為0
countDownLatch.await();
System.out.printf("%s時全部任務(wù)都完成,執(zhí)行合并計算.\n", new Date().toString());
executor.shutdown();
}
}
程序輸出:
Fri Aug 18 15:20:14 CST 2017時完成磁盤的統(tǒng)計任務(wù),耗費(fèi)0秒.
Fri Aug 18 15:20:15 CST 2017時完成磁盤的統(tǒng)計任務(wù),耗費(fèi)1秒.
Fri Aug 18 15:20:16 CST 2017時完成磁盤的統(tǒng)計任務(wù),耗費(fèi)2秒.
Fri Aug 18 15:20:17 CST 2017時完成磁盤的統(tǒng)計任務(wù),耗費(fèi)3秒.
Fri Aug 18 15:20:17 CST 2017時全部任務(wù)都完成,執(zhí)行合并計算.
從輸出中我們可以看出“執(zhí)行合并計算”這句是在所有線程完成統(tǒng)計任務(wù)之后才輸出的。
2、同步屏障CyclicBarrier
CyclicBarrier要做的事情是,讓一組線程到達(dá)一個屏障(也可以叫同步點(diǎn))時被阻塞,直到最后一個線程到達(dá)屏障時,屏障才會開門,所有被屏障攔截的線程才會繼續(xù)運(yùn)行。
CyclicBarrier初始化的時候,設(shè)置一個屏障數(shù)。線程調(diào)用await()方法的時候,這個線程就會被阻塞,當(dāng)調(diào)用await()的線程數(shù)量到達(dá)屏障數(shù)的時候,主線程就會取消所有被阻塞線程的狀態(tài)。
其構(gòu)造方法如下:
public CyclicBarrier(int parties){}
參數(shù)parties則為初始化時的屏障數(shù)
CyclicBarrier還提供一個更高級的構(gòu)造函數(shù)
public CyclicBarrier(int parties, Runnable barrierAction) {}
用于在線程到達(dá)屏障時,優(yōu)先執(zhí)行barrierAction,方便處理更復(fù)雜的業(yè)務(wù)場景
例如,用一個Excel保存了用戶所有的銀行流水,每個sheet保存一個賬戶近一年的每筆交易流水,現(xiàn)在需要統(tǒng)計用戶的日均交易流水,先用多線程處理每個sheet里的交易流水,都處理完后,得到每個sheet的日均交易流水,最后再用barrierAction用這些線程的計算結(jié)果,計算出整個Excel的日均銀行流水,代碼如下:
package com.dreyer.javadoc.thread;
import java.util.Map;
import java.util.concurrent.*;
/**
* @description 銀行交易流水服務(wù)類
* @author: 會跳舞的機(jī)器人
* @date: 16/5/15 上午11:29
*/
public class BankWaterService implements Runnable {
/**
* 創(chuàng)建4個屏障,處理完之后,執(zhí)行當(dāng)前類的run方法
*/
private CyclicBarrier cyclicBarrier = new CyclicBarrier(4, this);
/**
* 啟動4個線程
*/
private Executor executor = Executors.newFixedThreadPool(4);
/**
* 保存每個sheet計算出來的銀行交易流水結(jié)果
*/
private ConcurrentHashMap<String, Integer> sheetBankWaterCount = new ConcurrentHashMap<String, Integer>();
/**
* 交易流水統(tǒng)計
*/
private void count() {
for (int i = 0; i < 4; i++) {
executor.execute(new Runnable() {
public void run() {
// 模擬計算當(dāng)前sheet的銀行交易流水?dāng)?shù)據(jù)的業(yè)務(wù)處理
sheetBankWaterCount.put(Thread.currentThread().getName(), 1);
// 銀行交易流水計算完成后,插入一個屏障
try {
cyclicBarrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}
});
}
}
/**
* 匯總計算結(jié)果
*/
public void run() {
int result = 0;
for (Map.Entry<String, Integer> sheet : sheetBankWaterCount.entrySet()) {
result += sheet.getValue();
}
// 設(shè)置計算結(jié)果,并輸出
sheetBankWaterCount.put("result", result);
System.out.println(result);
}
public static void main(String[] args) {
BankWaterService service = new BankWaterService();
service.count();
}
}
3、控制并發(fā)線程數(shù)Semaphore
Semaphore被用于控制特定資源在同一個時間被訪問的線程數(shù)量,它通過協(xié)調(diào)各個線程,以保證資源可以被合理的使用。
做個比喻,把Semaphore比作是控制流量的紅綠燈,比如xx馬路要限制流量,只允許同時有一百輛車在馬路上行駛,其他的都必須在路口等待,所以前一百輛會看到綠燈,可以開進(jìn)馬路,后面的車會看到紅燈,不能開進(jìn)馬路,但是如果前面一百輛車中有5輛已經(jīng)離開了馬路,那后面就允許有5輛車駛?cè)腭R路,這里例子里說的車就是線程,駛?cè)腭R路就代表線程正在執(zhí)行,離開馬路就表示線程執(zhí)行完成,看到紅燈就代表線程被阻塞,不能執(zhí)行。
應(yīng)用場景:Semaph可以用來做流量限制,特別是公共資源有限的應(yīng)用場景,比如說數(shù)據(jù)庫連接。
假如有一個需求,要讀取幾萬個文件的數(shù)據(jù),因?yàn)槎际荌O密集型人物,我們可以啟動幾十個線程并發(fā)的讀取,但是如果讀取到內(nèi)存后,還需要儲存到數(shù)據(jù)庫,而數(shù)據(jù)庫的連接數(shù)只有10個,這時候我們就必須控制只有10個線程同時獲取到數(shù)據(jù)庫連接,否則會拋出異常提示無法連接數(shù)據(jù)庫。針對這種情況,我們就可以使用Semaphore來做流量控制。代碼如下:
package com.dreyer.javadoc.thread;
import java.util.concurrent.*;
/**
* @description
* @author: 會跳舞的機(jī)器人
* @date: 16/5/15 上午11:59
*/
public class SemaphoreDemo {
/**
* 線程數(shù)量
*/
private static final int THREAD_COUNT = 30;
/**
* 線程池
*/
private static ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
private static Semaphore semaphore = new Semaphore(10);
public static void main(String[] args) {
for (int i = 0; i < THREAD_COUNT; i++) {
executor.execute(new Runnable() {
public void run() {
try {
// 獲取一個"許可證"
semaphore.acquire();
// 模擬數(shù)據(jù)保存
TimeUnit.SECONDS.sleep(2);
System.out.println("save date...");
// 執(zhí)行完后,歸還"許可證"
semaphore.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
executor.shutdown();
}
}
在代碼中,雖然有30個線程在執(zhí)行,但是只運(yùn)行10個并發(fā)的執(zhí)行。所以我們可以看到在執(zhí)行的過程中
save data...是每10個輸出的。
Semaphore的構(gòu)造方法Semaphore(int permits)接受一個整形的數(shù)字,表示可用的許可證數(shù)量。
Semaphore(10)表示運(yùn)行10個線程獲取許可證,也就是最大的并發(fā)數(shù)是10。
Semaphore的用法也很簡單,首先使用Semaphore.acquire()方法獲取一個許可證,使用完之后調(diào)用release()方法歸還許可證。
4、線程間交換數(shù)據(jù)的Exchanger
Exchanger(交換者)是一個用于線程間協(xié)作的工具類。Exchanger用于進(jìn)行線程間的數(shù)據(jù)交換。它提供一個同步點(diǎn),在這個同步點(diǎn),兩個線程可以交換彼此的數(shù)據(jù)。這兩個線程通過exchange方法交換數(shù)據(jù),如果第一個線程先執(zhí)行exchange()方法,它會一直等待第二個線程也執(zhí)行exchange方法,當(dāng)兩個線程都到達(dá)同步點(diǎn)時,這兩個線程就可以交換數(shù)據(jù),將本線程生產(chǎn)出來的數(shù)據(jù)傳遞給對方。
下面來看一下Exchanger的應(yīng)用場景。
Exchanger可以用于遺傳算法,遺傳算法里需要選出兩個人作為交配對象,這時候會交換兩人的數(shù)據(jù),并使用交叉規(guī)則得出2個交配結(jié)果。Exchanger也可以用于校對工作,比如我們需要將紙制銀行流水通過人工的方式錄入成電子銀行流水,為了避免錯誤,采用AB崗兩人進(jìn)行錄入,錄入到Excel之后,系統(tǒng)需要加載這兩個Excel,并對兩個Excel數(shù)據(jù)進(jìn)行校對,看看是否
錄入一致,代碼示例如下:
package main.java.com.robot.demo;
import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @author: 會跳舞的機(jī)器人
* @date: 2017/8/18 15:32
* @description: 線程間交換數(shù)據(jù)的Exchanger示例
*/
public class ExchangerDemo {
private static final Exchanger<String> exchanger = new Exchanger<>();
/**
* 線程數(shù)
*/
private static ExecutorService executorService = Executors.newFixedThreadPool(2);
public static void main(String[] args) {
executorService.execute(new Runnable() {
@Override
public void run() {
try {
String a = "銀行流水A"; // A錄入的銀行流水?dāng)?shù)據(jù)
exchanger.exchange(a);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
executorService.execute(new Runnable() {
@Override
public void run() {
try {
String b = "銀行流水B"; // B錄入的銀行流水?dāng)?shù)據(jù)
String a = exchanger.exchange(b);
System.out.println("A和B的數(shù)據(jù)是否一致:" + a.equals(b) + ";A錄入的是" + a + ";B錄入的是:" + b);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
executorService.shutdown();
}
}
代碼輸出:
A和B的數(shù)據(jù)是否一致:false;A錄入的是銀行流水A;B錄入的是:銀行流水B
如果把變量a、b的值改為一樣的,則會輸出
A和B的數(shù)據(jù)是否一致:true;A錄入的是銀行流水A;B錄入的是:銀行流水A