JUC併發包基本使用

1、簡介

　　傳統的Java多線程開發中，wait、notify、synchronized等若是不注意使用的話，很容易引發死鎖、髒讀問題。Java1.5 版本開始增長 java.util.concurrent 併發編程包，簡化了多線程開發難度。添加了不少的多線程操做工具類，可根據實際需求去選擇使用。

　　JUC 經常使用工具類：

　　Semaphore - 信號量

　　ReentrantLock - 可重入鎖。以前有作過簡介使用，詳見 http://www.javashuo.com/article/p-tklmsesy-kk.html

　　ReadWriteLock - 讀寫鎖

　　BlockingQueue - 阻塞隊列。詳見 https://www.cnblogs.com/eric-fang/p/8989860.html

　　CountDownLatch - 計數器。在計數器歸零後，容許以前阻塞的若干線程繼續執行

　　CyclicBarrier - 柵欄。在某一條件達成以前，全部線程均阻塞等待

　　AtomicXXXXXXX - 原子操做類，常見的有：AtomicInteger、AtomicLong、AtomicBoolean。

　　TimeUnit - 時間枚舉類，提供一些時間的便捷操做

　　Executor、ExecutorService、Future ： 以前有作過簡介使用，詳見 http://www.javashuo.com/article/p-wgyamqkp-kh.html

2、使用

　　2.一、信號量Semaphore 

　　　　通常用於限定同時訪問操做的線程數量。例如：有時候能夠很好的限制公共資源的使用，例如若是開啓幾十個線程去讀取一些文件，而後讀取到的數據須要入庫的話，因爲數據庫的鏈接資源是稀缺資源，可能遠小於讀取文件的線程數，這時候能夠利用信號量去限制每次併發獲取數據庫鏈接資源的線程數。

　　　　以下示例代碼，雖然同時有10個線程執行，可是隻能容許2個線程的併發執行。

package com.cfang.prebo.thread;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class TestSemaphore2 {
    
    private static Semaphore semaphore = new Semaphore(2);
    
    private static ExecutorService executorService = Executors.newFixedThreadPool(10);

    public static void main(String[] args) {
        for(int i = 0; i < 10; i++) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        //申請通行證
                        semaphore.acquire();    
                        // 模擬業務邏輯
                        TimeUnit.SECONDS.sleep(2);
                        log.info("{} 處理完成", Thread.currentThread().getName());
                        //釋放通行證
                        semaphore.release();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        executorService.shutdown();
    }   
}

　　2.二、計數器CountDownLatch 

 　　　　同步計數器，構造方法傳值，用來限定計數器的次數。

　　　　countDown方法每次調用，計數器值減 1。CountDownLatch會一直阻塞着調用await方法的線程，直到計數器值變爲0。

package com.cfang.prebo.thread;

import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class TestCountDownLatch {

    private static CountDownLatch countDownLatch = new CountDownLatch(4);
    
    private static AtomicInteger integerVal = new AtomicInteger();
    
    public static void main(String[] args) throws Exception{
        for(int i = 0; i < 4; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    //業務處理邏輯
                    try {
                        int size = new Random().nextInt(100);
                        integerVal.getAndAdd(size);
                        TimeUnit.SECONDS.sleep(2);
                        log.info("{} 處理完成，{}", Thread.currentThread().getName(), size);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    countDownLatch.countDown();
                }
            }, "thread-" + i).start();
        }
        String threadName = Thread.currentThread().getName();
        log.info("{} thread waiting...", threadName);
        countDownLatch.await();
        log.info("{} doing, value: {}",threadName, integerVal.get());
    }
    
    
}

　　2.三、柵欄CyclicBarrier 

　　柵欄屏障，構造方法傳值來設定一個閾值。線程調用 await 方法的時候，線程就會被阻塞。當阻塞的線程數達到閾值的時候，全部阻塞線程所有放行。可重置重複使用。

package com.cfang.prebo.thread;

import java.util.Random;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class TestCyclicBarrier implements Runnable{

    private CyclicBarrier barrier = new CyclicBarrier(4, this);
    
    private static AtomicInteger integerVal = new AtomicInteger();
    
    public void count() {
        for(int i = 0; i < 4; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    //業務處理邏輯
                    try {
                        int size = new Random().nextInt(100);
                        integerVal.getAndAdd(size);
                        TimeUnit.SECONDS.sleep(2);
                        log.info("{} 處理完成，{}", Thread.currentThread().getName(), size);
                        barrier.await();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }, "thread-" + i).start();
        }
    }
    
    @Override
    public void run() {
        //業務邏輯處理完成後調用
        log.info("{} 統計完成，{}", Thread.currentThread().getName(), integerVal.get());
    }
    
    public static void main(String[] args) {
        TestCyclicBarrier testCyclicBarrier = new TestCyclicBarrier();
        testCyclicBarrier.count();
    }
}