所謂“工欲善其事，必先利其器”，有了這些併發工具，多線程控制變得小菜一碟了。
JDK中已經給我們內置了很多併發工具，都屬於應用類型，知道具體如何使用就好，主要講以下幾個類：
- CountDownLatch
- CyclicBarrier
- Semaphore
- LockSupport
- BlockingQueue
這次的幾個案例都需要實際運行，看運行效果才明白怎麼回事，代碼可以直接複製粘貼。
CountDownLatch
多線程控制類，計數器柵欄，當計數器滿足條件的時候，再開始執行接下來的操作。
public class CountDownLatchTest { static final int THREAD_COUNT = 10; static final CountDownLatch end = new CountDownLatch(THREAD_COUNT); public static void main(String[] args) throws InterruptedException { Runnable demo = new Runnable() { @Override public void run() { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("檢查完成"); end.countDown(); } }; //線程池內有5個線程方便看效果 ExecutorService executorService = Executors.newFixedThreadPool(5); for (int i = 0; i < THREAD_COUNT; i++) { executorService.submit(demo); } end.await(); System.out.println("一切就緒"); executorService.shutdown(); } CyclicBarrier
循環柵欄，可以看做CountDownLatch的重複利用。當滿足一定的條件時候，才開始執行某線程。
// 當線程的數量滿足條件時候，才開始執行。public class CyclicBarrierTest { public static void main(String[] args) { CyclicBarrier cyclicBarrier = new CyclicBarrier(4, new Runnable() { @Override public void run() { System.out.println("一切就緒，準備出發"); } }); Runnable task = new Runnable() { @Override public void run() { try { Thread.sleep(1000); System.out.println(Thread.currentThread().getId() + "：就緒"); cyclicBarrier.await(); } catch (InterruptedException e) { e.printStackTrace(); } catch (BrokenBarrierException e) { e.printStackTrace(); } } }; ExecutorService executorService = Executors.newFixedThreadPool(4); for (int i = 0; i < 4; i++) { executorService.submit(task); } executorService.shutdown(); }}
所有的線程都在等待，當等待的線程達到一定的數量，然後開始執行接下來的操作。
Semaphore
Semaphore，也是控制線程的一種手段，可以控制併發線程的數量，某些時候我們線程數過多，在訪問有限的資源時候，可以使用Semaphore來控制線程的數量。
public class SemaphoreDemo implements Runnable { Semaphore mSemaphore = new Semaphore(5); @Override public void run() { try { mSemaphore.acquire(); Thread.sleep(2000); System.out.println(Thread.currentThread().getId() + " done!"); mSemaphore.release(); } catch (InterruptedException e) { e.printStackTrace(); } } public static void main(String[] args) { ExecutorService executorService = Executors.newFixedThreadPool(20); SemaphoreDemo demo = new SemaphoreDemo(); for (int i = 0; i < 20; i++) { executorService.submit(demo); } executorService.shutdown(); ; }}
LockSupport
LockSupport提供了一些靜態方法用於阻塞線程，和喚醒線程的功能。
處於park()掛起狀態的線程是Waiting狀態，park()方法阻塞的線程還支持中斷，不拋出中斷異常的同時設置中斷標誌位，然後我們可以通過中斷標誌位來檢查。
public class LockDemo implements Runnable{ public static Object sObject = new Object(); @Override public void run() { synchronized (sObject){ System.out.println("當前線程名稱：" + Thread.currentThread().getName()); LockSupport.park(); if (Thread.currentThread().isInterrupted()){ System.out.println( Thread.currentThread().getName() + "被中斷了"); } System.out.println("執行結束"); } } public static void main(String[] args) throws InterruptedException { LockDemo demo = new LockDemo(); Thread t1 = new Thread(demo,"t1"); Thread t2 = new Thread(demo,"t2"); t1.start(); Thread.sleep(3000); t2.start(); t1.interrupt(); LockSupport.unpark(t2); }}
BlockingQueue
Java的Queue也是面試中經常提到的知識點，這次因為我們只涉及到併發相關知識，所以只提一些併發相關的Queue，關於Queue的具體分析等後面的數據結構系列的時候再詳細解說。
BlockingQueue是Java中的阻塞隊列，JDK中提供了7個阻塞隊列
- ArrayBlockingQueue : 數組實現的有界隊列，對元素進行FIFO(先進先出)的原則排序。
- LinkedBlockingQueue： 鏈表組成的有界隊列，長度默認最大值為Integer.MAX_VALUE，元素按FIFO原則排序，性能好於ArrayBlockingQueue。
- PriorityBlockingQueue：支持優先級的無界阻塞隊列。
- DelayQueue： 支持延遲獲取元素的無界阻塞隊列
- SynchronousQueue：不存儲元素的阻塞隊列。每一個put操作必須等待take操作，否則不能繼續添加元素。
- LinkedTransferQueue：鏈表組成的無界傳輸隊列
- LinkedBlockingDeque：由鏈表組成的雙向阻塞隊列。可以從兩段插入和移除元素。
帶大家看一下LinkedBlockingQueue的幾個關鍵方法：
//LinkedBlockingQueue 方法探索 // 添加元素 public boolean offer(E e) { if (e == null) throw new NullPointerException(); //如果隊列滿了，直接返回false final AtomicInteger count = this.count; if (count.get() == capacity) return false; // 創建新的節點 int c = -1; Node node = new Node(e); final ReentrantLock putLock = this.putLock; putLock.lock(); try { // 如果隊列不滿的話，就讓元素加入隊列。 //然後判斷，當前隊列元素是否滿了，不滿的話，通知notFull條件。 if (count.get() < capacity) { enqueue(node); c = count.getAndIncrement(); if (c + 1 < capacity) notFull.signal(); } } finally { putLock.unlock(); } // 假如添加的是第一個元素，通知隊列不為空了。 if (c == 0) signalNotEmpty(); return c >= 0; } public void put(E e) throws InterruptedException { if (e == null) throw new NullPointerException(); int c = -1; Node node = new Node(e); final ReentrantLock putLock = this.putLock; final AtomicInteger count = this.count; putLock.lockInterruptibly(); try { // 當隊列滿的時候進行等待。若不滿入隊 while (count.get() == capacity) { notFull.await(); } enqueue(node); c = count.getAndIncrement(); // 同offer if (c + 1 < capacity) notFull.signal(); } finally { putLock.unlock(); } // 同offer if (c == 0) signalNotEmpty(); } 可以看出添加元素上：
- 當隊列滿的時候，offer不添加元素，立刻返回。put則會阻塞操作，直到隊列為不滿。
- 還有一個帶參數的offer方法，和put的唯一區別就是有超時時間，在一段時間內隊列還不空的話，就返回。
//LinkedBlockingQueue 方法探索 // 移除 public E poll() { final AtomicInteger count = this.count; // 隊列為空，返回null if (count.get() == 0) return null; E x = null; int c = -1; final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { //隊列有元素的話，取出元素 //取出元素後如果隊列是不為空，發出不為空的信號。 if (count.get() > 0) { x = dequeue(); c = count.getAndDecrement(); if (c > 1) notEmpty.signal(); } } finally { takeLock.unlock(); } //如果取出元素之前，隊列是滿的，因為取出了元素，現在發出不滿的信號 if (c == capacity) signalNotFull(); return x; } public E take() throws InterruptedException { E x; int c = -1; final AtomicInteger count = this.count; final ReentrantLock takeLock = this.takeLock; takeLock.lockInterruptibly(); try { // 隊列為空的話，就等待隊列不為空。 while (count.get() == 0) { notEmpty.await(); } x = dequeue(); c = count.getAndDecrement(); if (c > 1) notEmpty.signal(); } finally { takeLock.unlock(); } if (c == capacity) signalNotFull(); return x; }可以看出LinkedBlockingQueue的移除操作poll和take方法：
- poll不阻塞，take會阻塞
- poll(long timeout, TimeUnit unit)，當隊列為空的時候，等待指定的時間，如果隊列扔為空，那麼就返回。
這次是帶領大家一起看了下LinkedBlockingQueue的關鍵方法，其它的隊列的操作也都類似，望大家自行查看，JDK中Queue的實現並不難理解。
最後
這次主要介紹了幾個併發中可能會用到的工具類，最後說了下JDK併發包中的阻塞隊列，阻塞隊列相對比較重要，就簡單的分析了其實現。
希望能幫助到大家。
參考
- 《Java併發實戰》
- 《Java高併發程序設計》
- 《併發編程的藝術》
鏈接：https://www.jianshu.com/p/8bdb65c2e525
來源：簡書
分享到:
閱讀更多 3T教育編程猿 的文章
