Java中的併發工具

在JDK的併發包裏面提供了幾個很是有用的併發工具，CountDownLatch、CyclicBarrier、Semaphore工具類提供了一種併發控制流程的手段。

1、CountDownLatch

CountDownLatch是一個同步輔助類，在完成一組正在其餘線程中執行的操做以前，它運行一個或者多個線程一直處於等待狀態。

CountDownLatch中有兩個關鍵的方法

public void countDown() {}

public boolean await(long timeout, TimeUnit unit){}

CountDownLatch是一個計數器，在它的構造方法中須要指定一個值，用來設定計數的次數。

每調用一次countDown()方法，數值便會減一，CountDownLatch會一直阻塞着調用await()方法的線程

直到計數器的值變爲0。

設想有這樣一個功能須要Thread一、Thread二、Thread三、Thread4四條線程分別統計C、D、E、F

四個盤的大小，全部線程都統計完畢交給主線程去作彙總，利用CountDownLatch來完成就很是輕鬆。

package com.dreyer.javadoc.thread;



import java.util.Date;

import java.util.Random;

import java.util.concurrent.*;



/**

 * @description CountDownLatch

 * @author: Dreyer

 * @date: 16/5/14 下午11:41

 */

public class CountDownLatchDemo {

    /**

     *

     */

    private static CountDownLatch countDownLatch = new CountDownLatch(4);



    /**

     * 線程池

     */

    private static ExecutorService executor = Executors.newFixedThreadPool(4);



    /**

     * 開啓的線程數

     */

    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 {

                        // 模擬業務邏輯的耗時

                        int timer = new Random().nextInt(5);

                        TimeUnit.SECONDS.sleep(timer);



                        System.out.printf("%s時完成磁盤的統計任務,耗費%d秒.\n", new Date().toString(), timer);

                        // 業務處理完成以後,計數器減一

                        countDownLatch.countDown();

                    } catch (InterruptedException e) {

                        e.printStackTrace();

                    }

                }

            });

        }

        // 主線程一直被阻塞,直到countDownLatch的值爲0

        countDownLatch.await();

        System.out.printf("%s時所有任務都完成,執行合併計算.\n", new Date().toString());

        executor.shutdown();

    }

}

2、CyclicBarrier

CyclicBarrier要作的事情是，讓一組線程到達一個屏障（也能夠叫同步點）時被阻塞，直到最後一個線程到達屏障時，屏障纔會開門，全部被屏障攔截的線程纔會繼續運行。

CyclicBarrier初始化的時候，設置一個屏障數。線程調用await()方法的時候，這個線程就會被阻塞，當調用await()的線程數量到達屏障數的時候，主線程就會取消全部被阻塞線程的狀態。

其構造方法以下：

public CyclicBarrier(int parties){}

參數parties則爲初始化時的屏障數

CyclicBarrier還提供一個更高級的構造函數

public CyclicBarrier(int parties, Runnable barrierAction) {}

用於在線程到達屏障時，優先執行barrierAction，方便處理更復雜的業務場景

例如，用一個Excel保存了用戶全部的銀行流水，每一個sheet保存一個帳戶近一年的每筆交易流水，如今須要統計用戶的日均交易流水，先用多線程處理每一個sheet裏的交易流水，都處理完後，獲得每一個sheet的日均交易流水，最後再用barrierAction用這些線程的計算結果，計算出整個Excel的日均銀行流水，代碼以下：

package com.dreyer.javadoc.thread;



import java.util.Map;

import java.util.concurrent.*;



/**

 * @description 銀行交易流水服務類

 * @author: Dreyer

 * @date: 16/5/15 上午11:29

 */

public class BankWaterService implements Runnable {

    /**

     * 建立4個屏障,處理完以後,執行當前類的run方法

     */

    private CyclicBarrier cyclicBarrier = new CyclicBarrier(4, this);



    /**

     * 啓動4個線程

     */

    private Executor executor = Executors.newFixedThreadPool(4);



    /**

     * 保存每一個sheet計算出來的銀行交易流水結果

     */

    private ConcurrentHashMap<String, Integer> sheetBankWaterCount = new ConcurrentHashMap<String, Integer>();



    /**

     * 交易流水統計

     */

    private void count() {

        for (int i = 0; i < 4; i++) {

            executor.execute(new Runnable() {

                public void run() {

                    // 模擬計算當前sheet的銀行交易流水數據的業務處理

                    sheetBankWaterCount.put(Thread.currentThread().getName(), 1);



                    // 銀行交易流水計算完成後,插入一個屏障

                    try {

                        cyclicBarrier.await();

                    } catch (InterruptedException e) {

                        e.printStackTrace();

                    } catch (BrokenBarrierException e) {

                        e.printStackTrace();

                    }

                }

            });

        }

    }



    /**

     * 彙總計算結果

     */

    public void run() {

        int result = 0;

        for (Map.Entry<String, Integer> sheet : sheetBankWaterCount.entrySet()) {

            result += sheet.getValue();

        }

        // 設置計算結果,並輸出

        sheetBankWaterCount.put("result", result);

        System.out.println(result);



    }

    public static void main(String[] args) {

        BankWaterService service = new BankWaterService();

        service.count();

    }

}

3、Semapphore

Semaphore被用於控制特定資源在同一個時間被訪問的線程數量，它經過協調各個線程，以保證資源能夠被合理的使用。

作個比喻，把Semaphore比做是控制流量的紅綠燈，好比xx馬路要如今流量，只容許同時有一百輛車在馬路上行駛，其餘的都必須在路口等待，因此前一百輛會看到綠燈，能夠開進馬路，後面的車會看到紅燈，不能開進馬路，可是若是前面一百輛車中有5輛已經離開了馬路，那後面就容許有5輛車駛入馬路，這裏例子裏說的車就是線程，駛入馬路就表明線程正在執行，離開馬路就表示線程執行完成，看到紅燈就表明線程被阻塞，不能執行。

應用場景

Semaph能夠用來作流量限制，特別是公共資源有限的應用場景，好比說數據庫鏈接。

假若有一個需求，要讀取幾萬個文件的數據，由於都是IO密集型人物，咱們能夠啓動幾十個線程併發的讀取，可是若是讀取到內存後，還須要儲存到數據庫，而數據庫的鏈接數只有10個，這時候咱們就必須控制只有10個線程同時獲取到數據庫鏈接，不然會拋出異常提示沒法鏈接數據庫。針對這種狀況，咱們就可使用Semaphore來作流量控制。代碼以下：

package com.dreyer.javadoc.thread;



import java.util.concurrent.*;



/**

 * @description

 * @author: Dreyer

 * @date: 16/5/15 上午11:59

 */

public class SemaphoreDemo {

    /**

     * 線程數量

     */

    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();



                        // 模擬數據保存

                        TimeUnit.SECONDS.sleep(2);

                        System.out.println("save date...");



                        // 執行完後,歸還"許可證"

                        semaphore.release();

                    } catch (InterruptedException e) {

                        e.printStackTrace();

                    }

                }

            });

        }

        executor.shutdown();

    }

}

在代碼中，雖然有30個線程在執行，可是隻運行10個併發的執行。因此咱們能夠看到在執行的過程當中

save data...是每10輸出的。

Semaphore的構造方法Semaphore(int permits)接受一個整形的數字，表示可用的許可證數量。

Semaphore(10)表示運行10個線程獲取許可證，也就是最大的併發數是10。

Semaphore的用法也很簡單，首先使用Semaphore.acquire()方法獲取一個許可證，使用完以後調用

release()方法歸還許可證。