Java 并发基础
Java 并發基礎
標簽 ： Java基礎
線程簡述
線程是進程的執行部分,用來完成一定的任務; 線程擁有自己的堆棧,程序計數器和自己的局部變量,但不擁有系統資源, 他與其他線程共享父進程的共享資源及部分運行時環境,因此編程時需要小心,確保線程不會妨礙同一進程中的其他線程;
- 多線程優勢
- 進程之間不能共享內存,但線程之間共享內存/文件描述符/進程狀態非常容易.
- 系統創建進程時需要為該其分配很多系統資源(如進程控制塊),但創建線程的開銷要小得多,因此線程實現多任務并發比進程效率高.
- Java語言內置多線程支持,而不是單純采用底層操作系統API調用, 從而可以簡化Java的多線程編程.
線程創建與啟動
Java使用java.lang.Thread代表線程(所有的線程對象必須是Thread類的實例).使用java.lang.Runnable java.util.concurrent.Callable和java.util.concurrent.Future來代表一段線程執行體(一段順序執行的代碼).一個線程的作用是完成一段程序流的執行,同時子線程的執行還可以跟父線程并行, 兩段線程的執行流程沒有關系, 父線程還可以繼續執行其他的事情.
繼承Thread
繼承Thread類,并重寫run()方法(代表線程執行體),然后調用start()方法來啟動線程.
/*** @author jifang* @since 16/1/20下午2:32.*/ public class ThreadStart {public static void main(String[] args) {new ConcreteThread().start();new ConcreteThread("second").start();for (int i = 0; i < 10; ++i) {System.out.println(Thread.currentThread().getName() + ": i");}}private static class ConcreteThread extends Thread {public ConcreteThread() {}public ConcreteThread(String name) {super(name);}@Overridepublic void run() {for (int i = 0; i < 10; ++i) {System.out.println(getName() + ": " + i);}}} }繼承Thread類來創建線程類時,多個線程之間無法共享線程類的實例變量.
實現Runnable
實現Runnable接口,重寫run()方法(同樣代表線程執行體),并將該類實例作為Thread的target提交給線程執行.
/*** @author jifang* @since 16/1/20下午2:47.*/ public class RunnableStart {public static void main(String[] args) {Runnable runnable = new ConcreteRunnable();new Thread(runnable, "first").start();new Thread(runnable).start();for (int i = 0; i < 10; ++i) {System.out.println(Thread.currentThread().getName() + " " + i);}}private static class ConcreteRunnable implements Runnable {private int i = 0;@Overridepublic void run() {for (; i < 10; ++i) {System.out.println(Thread.currentThread().getName() + " " + i);}}} }運行上例可以看到i值重復的現象,這是因為有多個線程都在修改同一個i值, 對于并發修改共享資源的情況,需要添加同步機制保護,詳見下面.
Runnable對象僅作為Thread對象的target,其包含的run()方法僅作為線程執行體.實際的線程對象依然是Thread實例, 只是該Thread線程執行的是target的run()方法.
Callable與Future
Callable接口提供一個call()方法作為線程執行體,相比于run(),call()可以有返回值,還可以聲明拋出異常.但它并不是Runnable接口的子接口, 所以不能直接作為target執行.因此Java又提供了Future接口來代表Callable中call()方法的返回值,并提供java.util.concurrent.FutureTask類實現Callable與Runnable接口(其實現了RunnableFuture接口,該接口同時繼承了Runnable Future),以作為Thread的target.
Future提供如下方法控制與其關聯的Callable:
| boolean cancel(boolean mayInterruptIfRunning) | Attempts to cancel execution of this task. |
| V get() | Waits if necessary for the computation to complete, and then retrieves its result. |
| V get(long timeout, TimeUnit unit) | Waits if necessary for at most the given time for the computation to complete, and then retrieves its result, if available. |
| boolean isCancelled() | Returns true if this task was cancelled before it completed normally. |
| boolean isDone() | Returns true if this task completed. |
Callable創建并啟動線程的步驟如下:
- 實現Callable接口并重寫call()方法;
- 使用FutureTask類包裝Callable對象;
- 將FutureTask實例提交給Thread并啟動新線程;
- 使用FutureTask的get()獲取子線程執行結束后的返回值.
由于實現Runnable和Callable的方式可以讓多個線程共享同一個target,因此適用于多個線程處理同一份資源的情況,從而將CPU/代碼/數據分開.
線程生命周期
當線程被new出并start后,他既不是馬上就進入執行狀態, 也不會一直處于執行狀態, 一個線程會經過新建NEW -> 就緒RUNNABLE -> 運行RUNNING -> 阻塞BLOCKED -> 死亡DEAD五種狀態切換.
1. 新建New
當new出一個Thread后,該線程處于新建狀態,此時他和其他Java對象一樣,僅由JVM為其分配內存.并沒有表現出任何線程的動態特征.
2. 就緒Runnable
當線程對象調用start()后,該線程處于就緒狀態,JVM會為其創建方法調用棧(Stack Trace)/線程控制塊/程序計數器(PC),處于這個狀態的線程表示是可以運行的.但何時運行,取決于JVM里線程調度器的調度.
3. 運行Running
如果處于就緒狀態的線程一旦獲得了CPU,就開始執行run()方法中的線程執行體,則線程進入運行狀態.
4. 阻塞Blocked
當發生如下情況時,線程會進入阻塞狀態
- 線程調用sleep()主動放棄處理器;
- 線程調用阻塞IO, 其IO資源未到;
- 線程試圖獲得同步監視器, 但同步監視器被其他線程持有;
- 線程等待某個通知wait();
- 調用了線程的suspend()方法(該方法將導致線程掛起,但這樣容易導致死鎖,不建議使用[詳細見線程同步]).
當前線程被阻塞之后, 其他線程就可以獲得執行的機會.
當發生如下情況, 線程可以解除阻塞, 重新進入就緒:
- 線程sleep()到達指定時間;
- 阻塞IO返回;
- 成功獲得同步監視器;
- 線程收到了其他線程發出的通知notify();
- 被suspend()的線程被調用了resume()恢復方法;
被阻塞的線程會在合適的時候重新進入就緒狀態.
5. 線程死亡
- run() / call()方法執行完成, 線程正常結束;
- 線程拋出未捕獲的Exception或Error;
- 直接調用線程的stop()方法結束該線程(該方法容易導致死鎖,不建議使用).
一旦子線程啟動起來后,就擁有和父線程相同的地位,不會受父線程的任何影響(因此當主線程結束時,其他線程不會同主線程一起結束).
為了測試某個線程是否生存, 可以調用Thread實例的isAlive()方法(就緒/運行/阻塞返回true, 新建/死亡返回false).
不要試圖對已經死亡的線程調用start()方法, 死亡線程將不可再次作為線程執行.否則會拋出java.lang.IllegalThreadStateException.
線程池
線程池會在系統啟動時即創建大量空閑線程,然后將一個Runnable/Callable對象提交給線程池,池就會分配/創建一個線程來執行他們的run()/call(),任務執行結束,該線程并不會死亡,而是再次返回池中變為空閑狀態,等待執行下一個任務;
線程池不僅可以避免每當有新任務就啟動一個新線程帶來的系統開銷,而且可以有效控制系統中并發線程的數量,一旦系統中的線程超過一定數量,將導致系統性能劇烈下降,甚至JVM崩潰,而線程池可以設置最大線程數以防止線程數超標.
Java提供java.util.concurrent.Executors工廠類來生產線程池, 該工廠類提供如下靜態方法:
| static ExecutorService newCachedThreadPool() | Creates a thread pool that creates new threads as needed, but will reuse previously constructed threads when they are available. |
| static ExecutorService newFixedThreadPool(int nThreads) | Creates a thread pool that reuses a fixed number of threads operating off a shared unbounded queue. |
| static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) | Creates a thread pool that can schedule commands to run after a given delay, or to execute periodically. |
| static ExecutorService newSingleThreadExecutor() | Creates an Executor that uses a single worker thread operating off an unbounded queue. |
| static ScheduledExecutorService newSingleThreadScheduledExecutor() | Creates a single-threaded executor that can schedule commands to run after a given delay, or to execute periodically. |
上面這些方法還有都有一個重載方法,需要使用java.util.concurrent.ThreadFactory參數,ThreadFactory是一個接口,用于自定義線程的創建策略.
1.java.util.concurrent.ExecutorService代表盡快執行任務的線程池,當有任務執行時,只需將RunnableCallable實例submit()給線程池就好(只池中有空閑線程,就立即執行任務),ExecutorService提供如下方法來提交任務:
| <T> Future<T> submit(Callable<T> task) | Submits a value-returning task for execution and returns a Future representing the pending results of the task. |
| Future<?> submit(Runnable task) | Submits a Runnable task for execution and returns a Future representing that task. |
| <T> Future<T> submit(Runnable task, T result) | Submits a Runnable task for execution and returns a Future representing that task. |
Java為ExecutorService提供了一個java.util.concurrent.ThreadPoolExecutor實現類,該類有如下構造方法:
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {// ... }因此, 如果默認的線程池策略(如最[小/大]線程數/線程等待時間)不能滿足我們的需求,我們可以自定義線程池策略.
2.ScheduledExecutorService線程池是ExecutorService的子接口,代表可以在指定延遲后或周期性執行線程任務.它提供了如下方法來提交任務:
| <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) |
| ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) |
| ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) |
| ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) |
其釋義可以參考JDK文檔;
/*** @author jifang* @since 16/1/20下午9:47.*/ public class ThreadPool {public static void main(String[] args) {ExecutorService pool = getThreadPool();pool.submit(new ConcreteRunnable());pool.submit(new ConcreteRunnable());pool.shutdown();}private static ExecutorService getThreadPool() {return Executors.newCachedThreadPool(new ThreadFactory() {@Overridepublic Thread newThread(Runnable r) {return new Thread(r);}});// return Executors.newCachedThreadPool();// return Executors.newFixedThreadPool(2);// return Executors.newSingleThreadExecutor();}private static class ConcreteRunnable implements Runnable {@Overridepublic void run() {for (int i = 0; i < 10; ++i) {try {Thread.sleep(10);} catch (InterruptedException e) {throw new RuntimeException(e);}System.out.println(Thread.currentThread().getName() + " " + i);}}}}- 使用自定義策略的線程池,提交Callable任務
用完一個線程池后, 應該調用該線程池的shutdown()方法,該方法將啟動線程池關閉序列,不再接收新任務,但會將以前所有已提交的任務盡快執行完成.所有任務都執行完,池中所有線程都會死亡.
線程控制
Java提供了一些工具方法,通過這些方法可以控制線程的執行.
Join
join()方法可以一個讓線程等待另一個線程執行完成: 調用線程被阻塞,知道被join()的線程執行完成.
該方法通常由主線程調用,將大問題劃分成小問題,每個小問題分配一個線程執行,當所有的小問題處理完成,再由主線程來做最后處理.如多線程排序,將一個大的排序任務分割為幾個小塊,分配給幾個線程,當所有子線程執行完成后,再由主線程進行歸并:
join()還其他重載形式,可以設定主調線程的最長等待時間.
后臺線程
后臺線程的任務是為其他線程提供服務,又被成為”守護線程”, JVM的垃圾回收線程就是典型的后臺守護線程.
調用Thread對象的setDaemon(true)方法可以將指定線程設置成后臺線程(在start()之前),isDaemon()可以判斷是否為后臺線程(主線程默認是非后臺線程, 非后臺線程創建的默認是非后臺線程, 反之亦然).
后臺線程的特征: 所有前臺線程死亡, 后臺線程會自動死亡.
Sleep
前面多次看到在線程的執行過程中調用sleep()讓線程睡眠(進入阻塞狀態),以模擬耗時的操作. 其方法簽名如下:
static void sleep(long millis) throws InterruptedException;由于sleep()會拋出CheckedException,因此可以將其包裝一下:
/*** @author jifang* @since 16/1/23 下午9:17.*/ public class SleepUtil {public static void sleep(long millis) {try {Thread.sleep(millis);} catch (InterruptedException e) {throw new RuntimeException(e);}} }sleep()還有重載形式, 但不常用.
Yield
yield()方法讓當前正在執行的線程暫停,但不是阻塞線程,而是讓該線程轉入就緒狀態,重新等待調度.
實際上,當某個線程調用yield()方法讓出處理器資源后,只有優先級與當前線程相同,或優先級比當前線程更高的處于就緒的線程才會獲得執行機會, 因此完全有可能線程轉入就緒后,調度器又將其調度出來重新執行.
注意: yield()方法可移植性并不是很好, 而且很有可能導致死鎖.所以并不推薦使用(詳細見線程同步).
線程優先級
每個線程都具有一定的優先級,優先級高的線程獲得更多的執行機會;默認情況下,每個子線程的優先級與子父線程相同(默認main線程具有普通優先級).
Thread類提供了setPriority(int newPriority)/getPriority()方法來設置/獲取線程優先級.newPriority的范圍為1~10,但由于這些級別需要操作系統的支持,但不同操作系統的優先級策略并不相同,因此推薦使用Thread類提供了三個靜態常量進行設置:
線程同步
模擬銀行取錢的場景,無線程同步:
- 賬戶
- 甲/乙線程取錢
運行如上程序, 很有可能會產生余額為負的情況(但現實中這是不可能存在的,出現這樣的結果就說明是我們的程序出錯了).
synchronized
之所以會出現上面的情況, 是因為run()方法不具有線程安全性(當賬戶余額為500時, 甲乙兩個線程的account.getBalance() > money都返回true(為了增大這類事件產生的概率,線程會在判斷完之后會sleep1毫秒以等待另一個線程),這樣兩個線程都會去取款300,因此會導致余額出現-100的情況).
為了解決該問題, Java多線程引入了同步監視器synchronized關鍵字;被synchronized保護的代碼稱之為同步代碼塊,線程開始執行同步代碼塊之前, 必須先獲得對同步監視器的鎖定.
任何時刻只能有一個線程獲得對同步監視器的鎖定,當同步代碼執行完后,該線程會自動釋放對同步監視器的鎖定.
@Override public void run() {while (true) {synchronized (account) {if (account.getBalance() > money) {System.out.println(Thread.currentThread().getName() + "取錢" + money + "成功");try {Thread.sleep(1);} catch (InterruptedException e) {throw new RuntimeException(e);}account.reduceBalance(money);System.out.println("\t" + Thread.currentThread().getName() + "成功后的余額: " + account.getBalance());} else {System.out.println(Thread.currentThread().getName() + "取錢失敗");System.out.println("\t" + Thread.currentThread().getName() + "失敗后的余額: " + account.getBalance());break;}}} }推薦使用可能被并發訪問的共享資源作為同步監視器.
synchronized關鍵詞還可以用于修飾方法,該方法稱之為同步方法,同步方法鎖定的同步監視器是this,也就是調用方法的對象.我們可將取錢的操作移動到Account中,并將該類改造成線程安全類:
/*** @author jifang* @since 16/1/21下午2:05.*/ public class Account {// ...public synchronized boolean draw(double money) {if (getBalance() > money) {System.out.println(Thread.currentThread().getName() + "取錢" + money + "成功");try {Thread.sleep(1);} catch (InterruptedException e) {throw new RuntimeException(e);}balance -= money;System.out.println("\t" + Thread.currentThread().getName() + "成功后的余額: " + getBalance());return true;} else {System.out.println(Thread.currentThread().getName() + "取錢失敗");System.out.println("\t" + Thread.currentThread().getName() + "失敗后的余額: " + getBalance());return false;}} }這樣Thread的run()方法則變得非常簡單:
@Overridepublic void run() {while (account.draw(money)) {}}synchronized可以修飾方法,也可以修改代碼塊,但不能修飾構造器,成員變量等.
同步監視器釋放
釋放同步監視器鎖定:
- 當前線程的同步方法/同步代碼塊執行結束, 釋放同步監視器;
- 當前線程在同步代碼塊/同步方法中遇到break/return終止該代碼塊/方法的執行, 釋放同步監視器.
- 當前線程在同步代碼塊/同步方法中出現了未處理的Error/Exception, 導致異常結束, 釋放同步監視器.
- 當前線程調用了同步對象的wait()方法,當前線程暫停,并釋放同步監視器.
不釋放同步監視器:
- 程序調用Thread.sleep()/Thread.yield()方法暫停當前線程執行.
- 其他線程調用當前線程的suspend()方法將線程掛起.
wait/notify
現在系統中有兩個線程,分別執行存錢/取錢,考慮這樣一種特殊的需求:”要求存錢者和取錢著不斷地重復存錢/取錢動作,同時規定不允許連續兩次存錢, 也不允許兩次連續取錢”.
可以借助Object類提供的wait()/notify()/notifyAll()三個方法來完成這一需求,但這三個方法必須由同步監視器對象來調用,因此可以分為以下兩種情況:
- 使用synchronized修飾的同步方法, 由于this就是同步監視器,所以可以在同步方法里面直接調用這三個方法.
- 使用synchronized修飾的同步代碼塊, 由于同步監視器是synchronized括號中的對象, 所以必須使用該對象調用這三個方法.
| void wait() | Causes the current thread to wait until another thread invokes the notify() method or the notifyAll() method for this object. (注意: 調用wait()方法的當前線程會釋放對同步監視器的鎖定) |
| void notify() | Wakes up a single thread that is waiting on this object’s monitor. |
| void notifyAll() | Wakes up all threads that are waiting on this object’s monitor. |
Lock
從1.5開始,Java提供了另外一種線程同步機制Lock,Lock提供比synchronized更廣泛的鎖定操作,并且支持多個相關的Condition.
java.util.concurrent.locks.Lock和java.util.concurrent.locks.ReadWriteLock是Java提供的兩類鎖的根接口,并且為Lock提供了ReentrantLock/ReentrantReadWriteLock.ReadLock/ReentrantReadWriteLock.WriteLock實現, 為ReadWriteLock提供ReentrantReadWriteLock實現.
Lock很容易實現對共享資源的互斥訪問:每次只能有一個線程對Lock加鎖,線程在訪問共享資源之前應先獲得Lock對象并lock(), 在訪問結束之后要unlock().
ReentrantLock表示可重入鎖,也就是說一個線程可以對已被加鎖的ReentrantLock鎖再次加鎖,ReentrantLock對象會維持一個計數器來追蹤lock()方法的嵌套調用,所以一段被鎖保護的代碼可以調用另一個被相同鎖保護的方法.
- 使用Lock實現生產者/消費者模式
使用Lock對象進行線程同步, 當加鎖/釋放鎖出現在不同的作用范圍時, 通常建議使用finally塊來確保一定會釋放鎖.
相對于Lock, synchronized則更加方便,可以避免很多鎖的常見編程錯誤,但Lock卻提供了synchronized所沒有功能,比如用于tryLock()嘗試加鎖, 獲取可中斷鎖的lockInterruptibly(), 還有在等待時間內獲取鎖的tryLock(long time, TimeUnit unit)方法.
Condition
使用Lock來保證互斥時,Java提供了一個java.util.concurrent.locks.Condition來保持協調,使用Condition可以讓那些已經得到Lock的線程無法繼續執行的而釋放Lock,也可以喚醒其他等待在該Condition上的線程.
Condition實例需要綁定在一個Lock對象上,使用Lock來保護Condition,因此調用Lock的newCondition()方法才能獲得Condition實例. 他提供了如下方法:
| void await() | Causes the current thread to wait until it is signalled or interrupted(當前線程會釋放對Lock的鎖定). |
| void signal() | Wakes up one waiting thread. |
| void signalAll() | Wakes up all waiting threads. |
下面使用Condition來模擬生產者/消費者模型(當倉庫中有產品時才能消費, 當倉庫未滿時才能生產):
public class Repository {/*使用mutex保護condition*/private final Lock mutex = new ReentrantLock();private final Condition addCondition = mutex.newCondition();private final Condition reduceCondition = mutex.newCondition();private int count;private int limit;public Repository(int count, int limit) {this.count = count;this.limit = limit;}private boolean canAdd(int count) {return this.count + count <= limit;}private boolean canReduce(int count) {return this.count - count >= 0;}public void add(int count) {try {// + 加鎖mutex.lock();while (!canAdd(count)) {System.out.printf("+... 生產者 %s is waiting...%n", Thread.currentThread().getName());addCondition.await();}this.count += count;System.out.printf("+ %s生產成功, 當前產品數%d%n", Thread.currentThread().getName(), getCount());// 喚醒消費線程reduceCondition.signalAll();} catch (InterruptedException e) {throw new RuntimeException(e);} finally {// + 解鎖mutex.unlock();}SleepUtil.sleep(80);}public void reduce(int count) {try {// - 加鎖mutex.lock();while (!canReduce(count)) {System.out.printf("-... 消費者 %s is waiting...%n", Thread.currentThread().getName());reduceCondition.await();}this.count -= count;System.out.printf("- %s消費成功, 當前產品數%d%n", Thread.currentThread().getName(), getCount());// 喚醒生產線程addCondition.signalAll();} catch (InterruptedException e) {throw new RuntimeException(e);} finally {// - 解鎖mutex.unlock();}SleepUtil.sleep(80);}private int getCount() {return count;} } public class ProducerConsumer {public static void main(String[] args) {ExecutorService pool = getExecutor();Repository repository = new Repository(0, 100);pool.submit(new Producer(repository, 30));pool.submit(new Producer(repository, 30));pool.submit(new Consumer(repository, 1));pool.submit(new Consumer(repository, 2));}private static ExecutorService getExecutor() {return new ThreadPoolExecutor(10, 20, 20, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());}private static class Producer implements Runnable {private Repository repository;private int produceCount;public Producer(Repository repository, int produceCount) {this.repository = repository;this.produceCount = produceCount;}@Overridepublic void run() {while (true) {repository.add(produceCount);}}}private static class Consumer implements Runnable {private Repository repository;private int consumeCount;public Consumer(Repository repository, int consumeCount) {this.repository = repository;this.consumeCount = consumeCount;}@Overridepublic void run() {while (true) {repository.reduce(consumeCount);}}} }ThreadLocal
java.lang.ThreadLocal和其他所有的同步機制一樣,都是為了解決多個線程中對同一共享變量(臨界資源)的訪問沖突問題,不過其他同步機制(synchronized/Lock)都是通過加鎖來實現多個線程對同一變量的安全訪問. 而ThreadLocal則從另外一個角度來解決這個問題– 將臨界資源進行復制(使其不再是臨界資源),每個線程都擁有一份,從而也就沒有必要再對其同步.
ThreadLocal代表一個線程局部變量,他為每一個使用該變量的線程都提供一個副本,使得每一個線程都可以獨立的改變自己的副本,而不會和其他的線程沖突(好像每個線程都完全擁有該變量).
ThreadLocal提供如下方法:
| T get() | Returns the value in the current thread’s copy of this thread-local variable. |
| void remove() | Removes the current thread’s value for this thread-local variable. |
| void set(T value) | Sets the current thread’s copy of this thread-local variable to the specified value. |
可以看到, 雖然在主線程里面對Account設置的初始值, 但是到子線程里面還是默認值.
建議: 如果多個線程之間需要共享資源, 以達到線程之間通信的目的, 就使用同步機制;如果僅僅需要隔離多個線程之間的共享沖突, 則可以選擇使用ThreadLocal.
線程通信
BlockingQueue
Java 1.5 提供了java.util.concurrent.BlockingQueue接口, 雖然他也是Queue的子接口, 但他的主要用途不是作為容器, 而是用作線程同步的工具 –多個線程通過交替向BlockingQueue放入/取出數據,從而完成線程間通信的目的.
BlockingQueue提供如下方法用于線程間通信:
Java為BlockingQueue提供了如下實現:
- ArrayBlockingQueue: 基于數組實現的BlockingQueue.
- LinkedBlockingQueue: 基于鏈表實現的BlockingQueue.
- SynchronousQueue: 同步隊列, 對隊列的存/取操作必須交替進行.
- PriorityBlockingQueue: 并不是標準的隊列,其take/poll/remove方法返回的是隊列中最小元素(其大小關系可以通過實現Comparator接口進行定制).
- DelayQueue: 一個特殊的BlockingQueue,底層基于PriorityBlockingQueue實現.DelayQueue要求集合元素必須實現Delayed接口(該接口里面只有一個long getDelay(TimeUnit unit)方法),DelayQueue根據集合元素的getDelay()方法的返回值進行排序.
使用BlockingQueue實現生產者/消費者模型
/*** @author jifang* @since 16/1/21下午9:11.*/ public class ConsumerProducer {private static final int CORE_POOL_SIZE = 5;private static final int MAX_POOL_SIZE = 20;private static final int KEEP_ALIVE_TIME = 20;public static void main(String[] args) {BlockingQueue<String> queue = getQueue();Producer producer = new Producer(queue, 1000);Consumer consumer = new Consumer(queue);ExecutorService pool = getThreadPool();pool.submit(producer);pool.submit(consumer);pool.submit(consumer);}private static BlockingQueue<String> getQueue() {return new ArrayBlockingQueue<>(10);}private static ExecutorService getThreadPool() {return new ThreadPoolExecutor(CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());}private static class Producer implements Runnable {private BlockingQueue<String> queue;private int count;public Producer(BlockingQueue<String> queue, int count) {this.queue = queue;this.count = count;}@Overridepublic void run() {Random random = new Random();while (count-- != 0) {try {Thread.sleep(100);queue.put("product " + random.nextInt(100));} catch (InterruptedException e) {throw new RuntimeException(e);}}}}private static class Consumer implements Runnable {private BlockingQueue<String> queue;public Consumer(BlockingQueue<String> queue) {this.queue = queue;}@Overridepublic void run() {while (true) {try {String product = queue.take();System.out.println(Thread.currentThread().getName() + " " + product);Thread.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}}}} }Disruptor
最近師兄技術分享提到了并發框架Disruptor,原因是最近在做數據遷移,需要多個線程之間進行數據通信,將原來老庫中的數據讀出,中間處理一道之后寫入新庫.第一次聽到它不免驚訝一番(竟然還有這么牛逼的東西(⊙﹏⊙)),于是就在網上找了幾篇文章,了解了一下他的應用場景/使用方法和基本原理,現炒現賣整理如下,詳細可參考我列在后面的參考文檔.
Disruptor入門示例
- 事件
事件(Event)就是通過 Disruptor 進行交換的數據類型
- 事件工廠
事件工廠(EventFactory)定義了如何初始化Disruptor的環形緩沖區RingBuffer
- 事件消費者
定義了消費者線程該如何處理事件Event
- 事件生產者
定義了生產者向RingBufferpush哪些數據,Disruptor的事件發布過程分為三個階段:
- 從RingBuffer中獲取下一個可以寫入的序列號sequence;
- 獲取sequence對應的事件對象Event,將數據寫入Event;
- 將填充好的Event提交到RingBuffer.
注意: buffer.publish(sequence)需放在finally中以確保一定會得到調用.如果某個請求的sequence未被提交,將會阻塞當前/其他Producer發布事件.
附: Disruptor還提供另一種發布事件的形式,簡化以上操作并確保publish()總是得到調用:
public class IntegerProducer {private RingBuffer<IntegerEvent> buffer;public IntegerProducer(RingBuffer<IntegerEvent> buffer) {this.buffer = buffer;}private EventTranslatorOneArg<IntegerEvent, Integer> translator = new EventTranslatorOneArg<IntegerEvent, Integer>() {@Overridepublic void translateTo(IntegerEvent event, long sequence, Integer value) {event.setValue(value);}};public void pushData(int value) {buffer.publishEvent(translator, value);} }Disruptor要求publish()必須得到調用(即使發生異常也要調用),那么就需要調用者在事件處理時判斷事件攜帶的數據是否是正確/完整的,而不是由Disruptor來保證.
- 啟動Disruptor
Disruptor選項
Disruptor可以根據不同的軟件/硬件來調整選項以獲得更高的性能. 其基本的選項有兩個:單生產者模式設定和可選的等待策略.
單生產者模式
在并發系統中提高性能的方式之一就是單一寫者原則:如果你的代碼中僅有一個事件生產者,那么可以在生成Disruptor時將其設置為單一生產者模式來提高系統的性能.
可選的等待策略
Disruptor定義了com.lmax.disruptor.WaitStrategy接口用于抽象Consumer如何等待新事件,其默認策略是BlockingWaitStrategy
這個策略的內部使用一個鎖Lock和條件變量Condition來控制線程的執行和等待.因此是最慢的等待策略(但也是CPU使用率最低和最穩定的策略).但Disruptor提供了根據不同的部署環境調整等待的方法(與單生產者模式一樣, 也是需要在構建Disruptor實例時指定), Disruptor提供了如下幾種常用策略.
| SleepingWaitStrategy | SleepingWaitStrategy方式是循環等待并且在循環中間調用LockSupport.parkNanos()來睡眠.它的優點在于生產線程只需要計數,而不執行任何指令;并且沒有條件變量的消耗,因此和BlockingWaitStrategy一樣,SleepingWaitStrategy的CPU使用率也比較低.但是,事件對象從生產者到消費者傳遞的延遲變大了.因此最好用在不需要低延遲,而事件發布對于生產者的影響比較小的情況下,如異步日志. |
| YieldingWaitStrategy | YieldingWaitStrategy是可以被用在低延遲系統中的兩個策略之一,這種策略在減低系統延遲的同時也會增加CPU運算量.該策略會循環等待sequence增加到合適值(循環中調用Thread.yield()以允許其他準備好的線程執行);如果需要高性能而且事件消費者線程比邏輯內核少,推薦使用YieldingWaitStrategy策略,如:在開啟超線程時. |
| BusySpinWaitStrategy | BusySpinWaitStrategy是性能最高的等待策略,同時也是對部署環境要求最高的策略(CPU空轉等待).這個策略最好用在要求極高性能且事件處理線數小于CPU邏輯核心數的場景中,如在禁用超線程技術時. |
我們指定單生產者模式和BusySpinWaitStrategy來啟動上例:
public class Client {// ...@Testpublic void client() {// 指定單生產者模式,與BusySpinWaitStrategy策略Disruptor<IntegerEvent> disruptor = new Disruptor<>(factory, ringBufferSize, executorService, ProducerType.SINGLE, new BusySpinWaitStrategy());//...} }當程序跑起來之后,可以查看機器的top輸出,直觀的感受一下Disruptor是多么消耗CPU…
線程安全集合
包裝不安全集合
平常使用的ArrayList LikedList HashMap HashSet LinkedHashMap LinkedHashSet TreeMap TreeSet等java.util包下的集合(除HashTable Vactor除外)都是線程不安全的, 當多個線程并發向這些集合中存/取數據時, 就可能會破獲這些集合的數據完整性.
因此java.util.Collections類提供了一些工具方法來將這些線程不安全的集合包裝成線程安全的集合.
| static <T> Collection<T> synchronizedCollection(Collection<T> c) | Returns a synchronized (thread-safe) collection backed by the specified collection. |
| static <T> List<T> synchronizedList(List<T> list) | Returns a synchronized (thread-safe) list backed by the specified list. |
| static <K,V> Map<K,V> synchronizedMap(Map<K,V> m) | Returns a synchronized (thread-safe) map backed by the specified map. |
| static <T> Set<T> synchronizedSet(Set<T> s) | Returns a synchronized (thread-safe) set backed by the specified set. |
| static <K,V> SortedMap<K,V> synchronizedSortedMap(SortedMap<K,V> m) | Returns a synchronized (thread-safe) sorted map backed by the specified sorted map. |
| static <T> SortedSet<T> synchronizedSortedSet(SortedSet<T> s) | Returns a synchronized (thread-safe) sorted set backed by the specified sorted set. |
注意: 如果需要將某個集合包裝成線程安全集合, 需要在創建之后立即包裝.
線程安全集合
從1.5開始, Java在java.util.concurrent包下提供了大量支持高效并發訪問的接口和實現.
從UML類圖可以看出,線程安全的集合類可分為以Concurrent開頭和以CopyOnWrite開頭的兩類.
以Concurrent開頭的集合代表了支持并發訪問的集合, 他們可以支持多個線程并發讀/寫, 這些寫入操作保證是線程安全的,而讀取操作不必鎖定, 因此在并發寫入時有較好的性能.
- ConcurrentLinkedQueue實現了多線程的高效訪問, 訪問時無需等待, 因此當多個線程共享訪問一個公共集合時, ConcurrentLinkedQueue是一個較好的選擇(不允許使用null元素).
- ConcurrentHashMap默認支持16個線程并發寫入(可以通過concurrencyLevel構造參數修改), 當線程數超過concurrencyLevel時, 可能有一些線程需要等待.
由于ConcurrentLinkedQueue ConcurrentHashMap支持并發訪問, 所以當使用迭代器來遍歷集合時, 該迭代器可能不能反映出創建迭代器之后所做的修改, 但程序不會拋出異常.
以CopyOnWrite開頭的集合采用了寫時復制技術.
- CopyOnWriteArrayList采用底層復制數組的方式來實現寫入操作: 當線程執行讀操作時, 線程會直接讀取集合本身, 無須加鎖和阻塞;但當線程對CopyOnWriteArrayList集合執行寫入操作(add/remove/set)時, 該集合會在底層復制一份新數組, 然后對新數組執行寫入操作.由于對CopyOnWriteArrayList的寫入是針對副本執行, 因此它是線程安全的.
注意: 由于CopyOnWriteArrayList的寫入操作需要頻繁的復制數組,因此寫入性能較差;但由于讀操作不用加鎖(不是同一個數組),因此讀操作非常快. 因此, CopyOnWriteArrayList適合在讀操作遠遠大于寫操作的場景中, 如緩存.
- 由于CopyOnWriteArraySet底層封裝的是CopyOnWriteArrayList, 因此他的實現機制完全類似于CopyOnWriteArrayList.
未處理異常
從1.5開始, Java增強了線程的異常處理功能,如果線程執行過程中拋出了一個未處理異常,JVM會在結束該線程之前查找是否有對應的java.lang.Thread.UncaughtExceptionHandler對象,如果找到了Handler對象, 則會調用該對象的void uncaughtException(Thread thread, Throwable exception)方法來處理該異常.
Thread提供了如下兩個方法來設置異常處理器:
| static void setDefaultUncaughtExceptionHandler(Thread.UncaughtExceptionHandler eh) | Set the default handler invoked when a thread abruptly terminates due to an uncaught exception, and no other handler has been defined for that thread. |
| void setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler eh) | Set the handler invoked when this thread abruptly terminates due to an uncaught exception. |
參考:
總結
- 上一篇： Exadata上的分页查询性能测试
- 下一篇： Java_Object类