2. 程式人生 > >java之JUC系列-外部Tools-Executors|Semaphor|Exchanger|CyclicBarrier|CountDownLatch

java之JUC系列-外部Tools-Executors|Semaphor|Exchanger|CyclicBarrier|CountDownLatch

阿新 • • 發佈：2018-12-23

前面寫了兩篇JDBC原始碼的文章，自己都覺得有點枯燥，先插一段JUC系列的文章來換換胃口，前面有文章大概介紹過J U C包含的東西，JUC體系包含的內容也是非常的多，不是一兩句可以說清楚的，我這首先列出將會列舉的JUC相關的內容，然後介紹本文的版本：Tools部分

J.U.C體系的主要大板塊包含內容，如下圖所示：


注意這個裡面每個部分都包含很多的類和處理器，而且是相互包含，相互引用的，相互實現的。

說到J UC其實就是說java的多執行緒等和鎖，前面說過一些狀態轉換，中斷等，我們今天來用它的tools來實現一些有些小意思的東西，講到其他內容的時候，再來想想這寫tools是怎麼實現的。

tools是本文說要講到的重點，而tools主要包含哪些東西呢：


Tools也包含了5個部分的知識：Executors、Semaphor、Exchanger、CyclicBarrier、CountDownLatch，其實也就是五個工具類，這5個工具類有神馬用途呢，就是我們接下來要將的內容了。

Executors：

其實它主要用來建立執行緒池，代理了執行緒池的建立，使得你的建立入口引數變得簡單，通過方法名便知道了你要建立的執行緒池是什麼樣一個執行緒池，功能大概是什麼樣的，其實執行緒池內部都是統一的方法來實現，通過構造方法過載，使得實現不同的功能，但是往往這種方式很多時候不知道具體入口引數的改變有什麼意思，除非讀了原始碼才知道，此時builder模式的方式來完成，builder什麼樣的東西它告訴你就可以。

常見的方法有(都是靜態方法)：

1、建立一個指定大小的執行緒池，如果超過大小，放入blocken佇列中，預設是LinkedBlockingQueue，預設的ThreadFactory為：Executors.defaultThreadFactory()，是一個Executors的一個內部類。

Executors.newFixedThreadPool(int) 

內部實現是：

    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
2、建立一個指定大小的執行緒池，如果超過大小，放入blocken佇列中,預設是LinkedBlockingQueue，自己指定ThreadFactory，自己寫的ThreadFactory，必須implements ThreadFactory，實現方法：newThread(Runnable)。

Executors.newFixedThreadPool(int,ThreadFactory) 
內部實現是：

    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>(),
                                      threadFactory);
    }

3、建立執行緒池長度為1的，也就是隻有一個長度的執行緒池，多餘的必須等待，它和呼叫Executors.newFixedThreadPool(1)得到的結果一樣：

 Executors.newSingleThreadExecutor() 

內部實現是：

    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
是不是蠻簡單的，就是在變引數，你自己也可以new的。

4、和方法3類似，可以自定義ThreadFactory，這裡就不多說了！

5、建立可以進行快取的執行緒池，預設快取60s，資料會放在一個SynchronousQueue上，而不會進入blocken佇列中，也就是隻要有執行緒進來就直接進入排程，這個不推薦使用，因為容易出問題，除非用來模擬一些併發的測試：

Executors.newCachedThreadPool();

內部實現為：

    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
6、和方法5類似，增加自定義ThreadFactory

7、新增一個Schedule的排程器的執行緒池，預設只有一個排程：

Executors.newSingleThreadScheduledExecutor();

內部實現為(這裡可以看到不是用ThreadPoolExector了，schedule換了一個類，內部實現通過ScheduledThreadPoolExecutor類裡面的內部類ScheduledFutureTask來實現的，這個內部類是private，預設是引用不到的哦)：

    public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
        return new DelegatedScheduledExecutorService
            (new ScheduledThreadPoolExecutor(1));
    }
8、和7一樣，增加自己定義的ThreadFactory

9、新增一個schedule的執行緒池排程器，和newFixedThreadPool有點類似：

Executors.newScheduledThreadPool();
內部程式碼為：

    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }

其實內部Exectors裡面還有一些其他的方法，我們就不多說明了，另外通過這裡，大家先可以瞭解一個大概，知道Exectors其實是一個工具類，提供一系列的靜態方法，來完成對對應執行緒池的形象化建立，所以不用覺得很神奇，神奇的是內部是如何實現的，本文我們不闡述文章中各種執行緒池的實現，只是大概上有個認識，等到我們專門將Exector系列的時候，我們會詳細描述這些細節。

OK，我們繼續下一個話題：

Semaphor，這個鳥東西是敢毛吃的呢？

答：通過名字就看出來了，是訊號量。

訊號量可以幹什麼呢？

答：根據一些閥值做訪問控制。

OK，我們這裡模擬一個當多個執行緒併發一段程式碼的時候，如何控制其訪問速度：

import java.util.Random;
import java.util.concurrent.Semaphore;


  public class SemaphoreTest {
      private final static Semaphore MAX_SEMA_PHORE = new Semaphore(10);
      public static void main(String []args) {
           for(int i = 0 ; i < 100 ; i++) {
                final int num = i;
                final Random radom = new Random();
                new Thread() {
                     public void run() {
                         boolean acquired = false;
                         try {
                              MAX_SEMA_PHORE.acquire();
                              acquired = true;
                              System.out.println("我是執行緒：" + num + " 我獲得了使用權！" + DateTimeUtil.getDateTime());
                              long time = 1000 * Math.max(1, Math.abs(radom.nextInt() % 10));
                              Thread.sleep(time);
                              System.out.println("我是執行緒：" + num + " 我執行完了！" + DateTimeUtil.getDateTime());
                         }catch(Exception e) {
                              e.printStackTrace();
                         }finally {
                              if(acquired) {
                                 MAX_SEMA_PHORE.release();
                              }
                         }
                      }
                }.start();
           }
      }
  }
這裡是簡單模擬併發100個執行緒去訪問一段程式，此時要控制最多同時執行的是10個，用到了這個訊號量，執行程式用了一個執行緒睡眠一個隨機的時間來代替，你可以看到後面有執行緒說自己釋放了，就有執行緒獲得了，沒釋放是獲取不到的，內部實現方面，我們暫時不管，暫時知道這樣用就OK。

接下來：

Exchanger十個神馬鬼東西呢？

答：執行緒之間互動資料，且在併發時候使用，兩兩交換，交換中不會因為執行緒多而混亂，傳送出去沒接收到會一直等，由互動器完成互動過程。

啥時候用，沒想到案例？

答：的確很少用，而且案例很少，不過的確有這種案例，Exchanger

import java.util.concurrent.Exchanger;

public class ExchangerTest {
	
	public static void main(String []args) {
		final Exchanger <Integer>exchanger = new Exchanger<Integer>();
		for(int i = 0 ; i < 10 ; i++) {
			final Integer num = i;
			new Thread() {
				public void run() {
					System.out.println("我是執行緒：Thread_" + this.getName() + "我的資料是：" + num);
					try {
						Integer exchangeNum = exchanger.exchange(num);
						Thread.sleep(1000);
						System.out.println("我是執行緒：Thread_" + this.getName() + "我原先的資料為：" + num + " , 交換後的資料為：" + exchangeNum);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}.start();
		}
	}
}


這裡執行你可以看到，如果某個執行緒和另一個執行緒傳送了資料，它接受到的資料必然是另一個執行緒傳遞給他的，中間步驟由Exchanger去控制，其實你可以說，我自己隨機取選擇，不過中間的演算法邏輯就要複雜一些了。

接下來：

CyclicBarrier，關卡模式，搞啥玩意的呢？

答：當你在很多環節需要卡住，要多個執行緒同時在這裡都達到後，再向下走，很有用途。

能否舉個例子，有點抽象？

答：團隊出去旅行，大家一起先達到酒店住宿，然後一起達到遊樂的地方遊玩，然後一起坐車回家，每次需要點名後確認相關人員均達到，然後LZ一聲令下，觸發，大夥就瘋子般的出發了。

下面的例子也是以旅遊的方式來呈現給大家：

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class BarrierTest {
	
	private static final int THREAD_COUNT = 10;
	
	private final static CyclicBarrier CYCLIC_BARRIER = new CyclicBarrier(THREAD_COUNT  ,
		new Runnable() {
			public void run() {
				System.out.println("======>我是導遊，本次點名結束，準備走下一個環節!");
			}
		}
	);
	
	public static void main(String []args) 
			throws InterruptedException, BrokenBarrierException {
		for(int i = 0 ; i < 10 ; i++) {
			new Thread(String.valueOf(i)) {
				public void run() {
					try {
						System.out.println("我是執行緒：" + this.getName() + " 我們達到旅遊地點！");
						CYCLIC_BARRIER.await();
						System.out.println("我是執行緒：" + this.getName() + " 我開始騎車！");
						CYCLIC_BARRIER.await();
						System.out.println("我是執行緒：" + this.getName() + " 我們開始爬山！");
						CYCLIC_BARRIER.await();
						System.out.println("我是執行緒：" + this.getName() + " 我們回賓館休息！");
						CYCLIC_BARRIER.await();
						System.out.println("我是執行緒：" + this.getName() + " 我們開始乘車回家！");
						CYCLIC_BARRIER.await();
						System.out.println("我是執行緒：" + this.getName() + " 我們到家了！");
					} catch (InterruptedException e) {
						e.printStackTrace();
					} catch (BrokenBarrierException e) {
						e.printStackTrace();
					}
				}
			}.start();
		}
	}
}

測試結果中可以發現，大家一起走到某個步驟後，導遊說：“我是導遊，本次點名結束，準備走下一個環節!”，然後才會進入下一個步驟，OK，這個有點意思吧，其實賽馬也是這個道理，只是賽馬通常只有一個步驟，所以我們還有一個方式是：

CountDownLatch的方式來完成賽馬操作，CountDownLatch是用計數器來做的，所以它不可以被複用，如果要多次使用，就要從新new一個出來才可以。我們下面的程式碼中，用兩組賽馬，每組5個參與者來，做一個簡單測試：

import java.util.concurrent.CountDownLatch;

public class CountDownLatchTest {
	
	private final static int GROUP_SIZE = 5;
	
	public static void main(String []args) {
		processOneGroup("分組1");
		processOneGroup("分組2");
	}
	
	private static void processOneGroup(final String groupName) {
		final CountDownLatch start_count_down = new CountDownLatch(1);
		final CountDownLatch end_count_down = new CountDownLatch(GROUP_SIZE);
		System.out.println("==========================>\n分組：" + groupName + "比賽開始：");
		for(int i = 0 ; i < GROUP_SIZE ; i++) {
			new Thread(String.valueOf(i)) {
				public void run() {
					System.out.println("我是執行緒組：【" + groupName + "】,第：" + this.getName() + " 號執行緒,我已經準備就緒！");
					try {
						start_count_down.await();//等待開始指令發出即：start_count_down.countDown();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					System.out.println("我是執行緒組：【" + groupName + "】,第：" + this.getName() + " 號執行緒,我已執行完成！");
					end_count_down.countDown();
				}
			}.start();
		}
		try {
			Thread.sleep(1000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println("各就各位，預備！");
		start_count_down.countDown();//開始賽跑
		try {
			end_count_down.await();//等待多個賽跑者逐個結束
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		System.out.println("分組：" + groupName + "比賽結束！");
	}
}
有點意思哈，如果你自己用多執行緒實現是不是有點麻煩，不過你可以用Thread的join方法來實現，也就是執行緒的發生join的時候，當前執行緒（一般是主執行緒）要等到對應執行緒執行完run方法後才會進入下一步，為了模擬下，我們也來玩玩： 
public class ThreadJoinTest {
	
	private final static int GROUP_SIZE = 5;

	public static void main(String []args) throws InterruptedException {
		Thread []threadGroup1 = new Thread[5];
		Thread []threadGroup2 = new Thread[5];
		for(int i = 0 ; i < GROUP_SIZE ; i++) {
			final int num = i;
			threadGroup1[i] = new Thread() {
				public void run() {
					int j = 0;
					while(j++ < 10) {
						System.out.println("我是1號組執行緒：" + num + " 這個是我第：" + j + " 次執行！");
					}
				}
			};
			threadGroup2[i] = new Thread() {
				public void run() {
					int j = 0;
					while(j++ < 10) {
						System.out.println("我是2號組執行緒：" + num + " 這個是我第：" + j + " 次執行！");
					}
				}
			};
			threadGroup1[i].start();
		}
		for(int i = 0 ; i < GROUP_SIZE ; i++) {
			threadGroup1[i].join();
		}
		System.out.println("-==================>執行緒組1執行完了，該輪到俺了！");
		for(int i = 0 ; i < GROUP_SIZE ; i++) {
			threadGroup2[i].start();
		}
		for(int i = 0 ; i < GROUP_SIZE ; i++) {
			threadGroup2[i].join();
		}
		System.out.println("全部結束啦！哈哈，回家喝稀飯！");
	}
}


程式碼是不是繁雜了不少，呵呵，我們再看看上面的訊號量，如果不用工具，自己寫會咋寫，我們模擬CAS鎖，使用Atomic配合完成咋來做呢。也來玩玩，呵呵： 
import java.util.concurrent.atomic.AtomicInteger;

public class ThreadWaitNotify {
	
	private final static int THREAD_COUNT = 100;
	
	private final static int QUERY_MAX_LENGTH = 2;
	
	private final static AtomicInteger NOW_CALL_COUNT = new AtomicInteger(0);

	public static void main(String []args) throws InterruptedException {
		Thread []threads = new Thread[THREAD_COUNT];
		for(int i = 0 ; i < THREAD_COUNT ; i++) {
			threads[i] = new Thread(String.valueOf(i)) {
				synchronized public void run() {
					int nowValue = NOW_CALL_COUNT.get();
					while(true) {
						if(nowValue < QUERY_MAX_LENGTH && NOW_CALL_COUNT.compareAndSet(nowValue, nowValue + 1)) {
							break;//獲取到了
						}
						try {
							this.wait(1000);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
						nowValue = NOW_CALL_COUNT.get();//獲取一個數據，用於對比
					}
					System.out.println(this.getName() + "======我開始做操作了！");
					try {
						Thread.sleep(1000);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					System.out.println(this.getName() + "======操作結束了！");
					NOW_CALL_COUNT.getAndDecrement();
					this.notify();
				}
			};
		}
		for(int i = 0 ; i < THREAD_COUNT ; i++) {
			threads[i].start();
		}
	}
}

還是有點意思哈，這樣寫就是大部分人對while迴圈那部分會寫暈掉，主要是要不斷去判定和嘗試，wait()預設是長期等待，但是我們不想讓他長期等待，就等1s然後再嘗試，其例項子還可以改成wait一個隨機的時間範圍，這樣模擬的效果會更加好一些；另外實際的程式碼中，如果獲取到鎖後，notify方法應當放在finally中，才能保證他肯定會執行notify這個方法。

OK，本文就是用，玩，希望玩得有點爽，我們後面會逐步介紹它的實現機制以及一寫執行緒裡頭很好用，但是大家又不是經常用的東西。

相關推薦

java之JUC系列-外部Tools-Executors|Semaphor|Exchanger|CyclicBarrier|CountDownLatch

前面寫了兩篇JDBC原始碼的文章，自己都覺得有點枯燥，先插一段JUC系列的文章來換換胃口，前面有文章大概介紹過J U C包含的東西，JUC體系包含的內容也是非常的多，不是一兩句可以說清楚的，我這首先列出將會列舉的JUC相關的內容，然後介紹本文的版本：Tools部分 J.U.

Java之道系列：BigDecimal如何解決浮點數精度問題

IEEE 754 IEEE二進位制浮點數算術標準（IEEE 754）是20世紀80年代以來最廣泛使用的浮點數運算標準，為許多CPU與浮點運算器所採用。這個標準定義了表示浮點數的格式（包括負零-0）與反常值（denormal number）），一些

Java JUC之Atomic系列12大類例項講解和原理分解

在java6以後我們不但接觸到了Lock相關的鎖，也接觸到了很多更加樂觀的原子修改操作，也就是在修改時我們只需要保證它的那個瞬間是安全的即可，經過相應的包裝後可以再處理物件的併發修改，以及併發中的ABA問題，本文講述Atomic系列的類的實現以及使用方法，其中包含： 基本類

java設計模式系列之設計模式概要（1）

而不是 行為型模式 一句話 創建 rom 多次 ati 代理模式 之間 一、什麽是設計模式 　　設計模式（Design pattern）是一套被反復使用、多數人知曉的、經過分類編目的、代碼設計經驗的總結。使用設計模式是為了可重用代碼、讓代碼更容易被他人理解、保證代碼可靠性。

Java命令學習系列之jmap

.com respond 標識 config 一個bug running vivo huang 父類 jmap是JDK自帶的工具軟件，主要用於打印指定Java進程(或核心文件、遠程調試服務器)的共享對象內存映射或堆內存細節。可以使用jmap生成Heap Dump。 什麽

Java命令學習系列之jstack

數據庫 get image for 線程調度 nbsp 額外 無限 其他 jstack是java虛擬機自帶的一種堆棧跟蹤工具。 功能 jstack用於生成java虛擬機當前時刻的線程快照。線程快照是當前java虛擬機內每一條線程正在執行的方法堆棧的集合，生成線程快照的主

Java並發系列[1]----AbstractQueuedSynchronizer源碼分析之概要分析

人的 等等 子類 運行 pos unsafe fse oid 出了 學習Java並發編程不得不去了解一下java.util.concurrent這個包，這個包下面有許多我們經常用到的並發工具類，例如：ReentrantLock, CountDownLatch, Cyclic

Java並發系列[2]----AbstractQueuedSynchronizer源碼分析之獨占模式

cancel get color 中斷方式 方法 來看 .cn syn .com 在上一篇《Java並發系列[1]----AbstractQueuedSynchronizer源碼分析之概要分析》中我們介紹了AbstractQueuedSynchronizer基本的一些概念，

Java虛擬機系列之Java內存結構簡介

內存空間 指示器 私有 以及 並且 內存區域 在服務器 規範 基礎 本文我們將講解Java虛擬機中各個區域以及各個區域的作用。 一.程序計數器什麽是程序計數器，有什麽作用？程序技術器是一塊比較小的內存區域，主要當做是線程中所執行的字節碼的行號指示器，字節碼解釋器工作時就是通

JAVA小白系列之第二個分支：面向物件程式設計

想必剛入門的小白總是搞不清楚面向物件和麵向過程這兩個概念，雖然我在前面有所提及，但是都是泛泛而談，也有各種語言融入進去，不理解的會越來越暈，那麼，這節我就專門來講講什麼是JAVA的面向物件。 面向過程和麵向物件回顧 不記得概念的可以具體看看之前的文章，這裡僅做簡單的回顧： 面向過程 關注於流

【紮實基本功】Java基礎教程系列之多執行緒

1. 多執行緒的概念 1.1 程序、執行緒、多程序的概念 程序：正在進行中的程式（直譯）。 執行緒是程式執行的一條路徑, 一個程序中可以包含多條執行緒。 一個應用程式可以理解成就是一個程序。 多執行緒併發執行可以提高程式的效率, 可以同時完成多項工作。 1.

Java工程師之SpringBoot系列教程前言&目錄

前言 與時俱進是每一個程式設計師都應該有的意識，當一個Java程式設計師在當代步遍佈的時候，你就行該想到我能多學點什麼。可觀的是後端的框架是穩定的，它們能夠維持更久的時間在應用中，而不用擔心技術的更新換代。但是類似SSH,SSM這些框架已經太久了，人們迫不及待地想使用更為優雅而又簡便的框架來代替，所以Spr

Java 併發程式設計系列之閉鎖（CountDownLatch）

在講閉鎖之前，我們先來思考一個問題：在多執行緒環境下，主執行緒列印一句話，如何保證這句話最後（其他執行緒全部執行完畢）列印？ 博主目前可以想到的實現方式有兩種。一種是通過 join() 方法實現，另一

Java 併發程式設計系列之帶你瞭解多執行緒

早期的計算機不包含作業系統，它們從頭到尾執行一個程式，這個程式可以訪問計算機中的所有資源。在這種情況下，每次都只能執行一個程式，對於昂貴的計算機資源來說是一種嚴重的浪費。 作業系統出現後，計算機可以執行多個程式，不同的程式在單獨的程序中執行。作業系統負責為各個獨

Java開發之上帝之眼系列教程前言和章節目錄彙總

前言 如果您正在為Java後端龐大的體系所困擾，如果您正在為各種繁出不窮的技術和各種框架所迷茫，那麼本系列文章將帶您窺探Java龐大的體系。本系列教程希望您能站在上帝的角度去觀察（瞭解）Java體系。

Java上帝之眼系列Spring框架裝配Bean的三種方式

歡迎檢視Java開發之上帝之眼系列教程，如果您正在為Java後端龐大的體系所困擾，如果您正在為各種繁出不窮的技術和各種框架所迷茫，那麼本系列文章將帶您窺探Java龐大的體系。本系列教程希望您能站在上帝的角度去觀察（瞭解）Java體系。使Java的各種後端技術在

Java上帝之眼系列配置Spring專案檔案上傳兩種方式（全解析）

歡迎檢視Java開發之上帝之眼系列教程，如果您正在為Java後端龐大的體系所困擾，如果您正在為各種繁出不窮的技術和各種框架所迷茫，那麼本系列文章將帶您窺探Java龐大的體系。本系列教程希望您能站在上帝

BAT常見的Java面試題系列之集合類

一、ArrayList和Vector的區別？如何理解Vector的執行緒安全？ 共同點 1、兩者都實現了List介面，並且屬於有序儲存集合，可根據索引直接取值。 2、允許集合內的元素重複或為Null。 區別 1、ArrayList是執行緒不安全的，Vector是執行

JAVA小白系列之第五個分支：LinkedList容器深入分析一

上一節我們瀏覽了ArrayList容器，來總結一下？ 儲存資料的為一個名叫做elementData的陣列預設容量為10 擴容大小為原容量的一半即originSize+originSize>>2的大小 擴容方式為建立新的陣列並且通過陣列的複製來完成擴容 刪除

Java併發程式設計系列之十五 Executor框架

                     Java使用執行緒完成非同步任務是很普遍的事，而執行緒的建立與銷燬需要一定的開銷，如果每個任務都需要建立一個執行緒將會消耗大量的計算資源，JDK 5之後把工作單元和執行機制區分開了，工作單元包括Runnable和Callable，而執行機制則由Executor框架提供