Java多線程入門(mén)不完全指南
序言
最近在讀《把時(shí)間當(dāng)作朋友》，序言就教導(dǎo)我“無(wú)論是誰(shuí)，都是在某一刻意識(shí)到時(shí)間的珍貴，并且注定會(huì)因懂事太晚而多少有些后悔”。想想我，快走出奔三的泥沼，踏入奔四的深淵，越來(lái)越意識(shí)到時(shí)間的彌足珍貴。于是就想著，趁著還有些精力的時(shí)候，記錄一下所聞、所學(xué)、所感。思考良久，記錄些什么呢？先記錄一下我這拙劣而又不放棄的Java學(xué)習(xí)之路吧,挖坑一篇"粗而廣"的Java多線程介紹壓壓驚。
基礎(chǔ)知識(shí)
線程與進(jìn)程
- 線程：是操作系統(tǒng)能夠進(jìn)行運(yùn)算調(diào)度的最小單位，是進(jìn)程中的實(shí)際運(yùn)作單位。一條線程是進(jìn)程中一個(gè)單一順序的控制流。
- 進(jìn)程：是計(jì)算機(jī)中已運(yùn)行程序的實(shí)體，是線程的容器。
同步與異步
- 同步：在發(fā)出一個(gè)調(diào)用時(shí)，在沒(méi)有得到結(jié)果之前，該調(diào)用就不返回。一旦返回，必然會(huì)得到返回值。
- 異步：在調(diào)用發(fā)出之后，這個(gè)調(diào)用就直接返回。隨后，被調(diào)用者通過(guò)狀態(tài)、通知來(lái)通知調(diào)用者，或者通過(guò)回調(diào)函數(shù)來(lái)處理調(diào)用。
同步與異步關(guān)注的是消息通信機(jī)制。
阻塞與非阻塞
- 阻塞：調(diào)用結(jié)果返回之前，當(dāng)前線程會(huì)被掛起。調(diào)用線程只有在得到結(jié)果之后才會(huì)返回。
- 非阻塞：調(diào)用在不能立刻得到結(jié)果之前，該調(diào)用不會(huì)阻塞當(dāng)前線程，當(dāng)前線程仍會(huì)處理其他調(diào)用。
阻塞與非阻塞關(guān)注的是程序在等待調(diào)用結(jié)果（消息，返回值）時(shí)的狀態(tài)。
并行與并發(fā)
- 并發(fā)：在同一個(gè)處理器上“同時(shí)”處理多個(gè)任務(wù)。通過(guò)cpu調(diào)度算法，讓用戶看上去是同時(shí)處理。
- 并行：在多臺(tái)機(jī)器上同時(shí)處理多個(gè)任務(wù)，真正的同時(shí)。
多線程與線程安全
- 多線程:一個(gè)程序?qū)崿F(xiàn)多個(gè)線程并發(fā)執(zhí)行。
- 線程安全:多個(gè)線程同時(shí)執(zhí)行同一段代碼，線程的調(diào)度順序不會(huì)影響該段代碼的任何結(jié)果。線程安全主要通過(guò)線程同步實(shí)現(xiàn)。
線程的狀態(tài)
5種狀態(tài)
根據(jù)線程的生命周期，可以將線程分為以下5種狀態(tài)：
- NEW(新建)：創(chuàng)建了一個(gè)線程，尚未啟動(dòng)
- RUNNABLE(可運(yùn)行)：線程創(chuàng)建后，其他線程調(diào)用了該線程的
start()方法，該線程便等待被CPU調(diào)度執(zhí)行。 - RUNNING（運(yùn)行）：RUNNABLE狀態(tài)的線程被CPU調(diào)度，獲取了CPU時(shí)間片，運(yùn)行
run()方法。 - BLOCKERD（阻塞）：線程無(wú)法獲取CPU時(shí)間片，暫時(shí)停止運(yùn)行的狀態(tài)。這個(gè)狀態(tài)的線程只有狀態(tài)轉(zhuǎn)為RUNNABLE時(shí)，才有機(jī)會(huì)被CPU調(diào)度執(zhí)行。阻塞有三種情況：
- 等待阻塞：如RUNNING狀態(tài)的線程執(zhí)行
wait()方法。 - 同步阻塞：如RUNNING狀態(tài)的線程在獲取同步鎖時(shí)，同步鎖被其他線程占用。
- 其他阻塞：如RUNNING狀態(tài)的線程執(zhí)行
sleep()方法或其他線程調(diào)用join()方法。
- 等待阻塞：如RUNNING狀態(tài)的線程執(zhí)行
- DEAD(死亡)：線程
run()方法執(zhí)行結(jié)束或異常退出，該線程死亡，結(jié)束生命周期。
狀態(tài)轉(zhuǎn)換
等待隊(duì)列和鎖池是如何工作的？
Thread類(lèi)定義的線程狀態(tài)
-
NEW(新建)：線程創(chuàng)建后未啟動(dòng)。
/** * Thread state for a thread which has not yet started. */ -
RUNNABLE(可運(yùn)行)：等待系統(tǒng)資源運(yùn)行的線程。
/** * Thread state for a runnable thread. A thread in the runnable * state is executing in the Java virtual machine but it may * be waiting for other resources from the operating system * such as processor. */ -
BLOCKED(阻塞)：當(dāng)一個(gè)線程要進(jìn)入synchronized語(yǔ)句塊/方法時(shí)，如果沒(méi)有獲取到鎖，會(huì)變成BLOCKED?；蛘咴谡{(diào)用Object.wait()后，被notify()喚醒，再次進(jìn)入synchronized語(yǔ)句塊/方法時(shí)，如果沒(méi)有獲取到鎖，會(huì)變成BLOCKED。進(jìn)入阻塞狀態(tài)是被動(dòng)的。
/** * Thread state for a thread blocked waiting for a monitor lock. * A thread in the blocked state is waiting for a monitor lock * to enter a synchronized block/method or * reenter a synchronized block/method after calling * {@link Object#wait() Object.wait}. */ -
WAITING(無(wú)限等待)：等待其它線程執(zhí)行后，顯示喚醒。調(diào)用鎖對(duì)象的
wait()方法并未設(shè)置時(shí)間、其它線程調(diào)用join()方法并未設(shè)置時(shí)間、調(diào)用LockSupport.park()方法都會(huì)使當(dāng)前線程進(jìn)入此狀態(tài)。進(jìn)入等待狀態(tài)是主動(dòng)的。/** * Thread state for a waiting thread. * A thread is in the waiting state due to calling one of the * following methods: * <ul> * <li>{@link Object#wait() Object.wait} with no timeout</li> * <li>{@link #join() Thread.join} with no timeout</li> * <li>{@link LockSupport#park() LockSupport.park}</li> * </ul> * * <p>A thread in the waiting state is waiting for another thread to * perform a particular action. * * For example, a thread that has called <tt>Object.wait()</tt> * on an object is waiting for another thread to call * <tt>Object.notify()</tt> or <tt>Object.notifyAll()</tt> on * that object. A thread that has called <tt>Thread.join()</tt> * is waiting for a specified thread to terminate. */ -
TIMED_WAITING(限期等待)：等待一段時(shí)間后被系統(tǒng)喚醒，不需要顯示被喚醒。調(diào)用
Thread.sleep()方法、調(diào)用鎖對(duì)象的wait()方法并設(shè)置時(shí)間、其它線程調(diào)用join()方法并設(shè)置時(shí)間、調(diào)用LockSupport.parkNanos()方法、調(diào)用LockSupport.parkUntil()方法都會(huì)使線程進(jìn)入此狀態(tài)。/** * Thread state for a waiting thread with a specified waiting time. * A thread is in the timed waiting state due to calling one of * the following methods with a specified positive waiting time: * <ul> * <li>{@link #sleep Thread.sleep}</li> * <li>{@link Object#wait(long) Object.wait} with timeout</li> * <li>{@link #join(long) Thread.join} with timeout</li> * <li>{@link LockSupport#parkNanos LockSupport.parkNanos}</li> * <li>{@link LockSupport#parkUntil LockSupport.parkUntil}</li> * </ul> */ -
TERMINATED(結(jié)束)：線程
run()方法執(zhí)行結(jié)束或異常退出后的線程狀態(tài)。/** * Thread state for a terminated thread. * The thread has completed execution. */
VisualVM線程監(jiān)控
通過(guò)JDK自帶的VisualVM工具可以監(jiān)控線程的運(yùn)行狀態(tài)，按照下圖操作可以抓取線程Dump，監(jiān)控線程的狀態(tài)。
VisualVM將線程的狀態(tài)分為五種：運(yùn)行、休眠、等待、駐留、監(jiān)視，與Thread類(lèi)中的線程狀態(tài)對(duì)應(yīng)如下：
| Thread類(lèi) | VisualVM |
|---|---|
| RUNNABLE | 運(yùn)行 |
| TIMED_WAITING (sleeping) | 休眠 |
| TIMED_WAITING (on object monitor) WAITING (on object monitor) | 等待 |
| TIMED_WAITING (parking) WAITING (parking) | 駐留 |
| BLOCKED (on object monitor) | 監(jiān)視 |
內(nèi)存原子性、可見(jiàn)性和有序性
Java內(nèi)存模型
概念
- 線程之間通信由Java內(nèi)存模型控制，內(nèi)存模型決定了一個(gè)線程對(duì)共享變量的寫(xiě)入何時(shí)對(duì)另一個(gè)線程可見(jiàn)。
- 每個(gè)線程都被抽象出一個(gè)保存共享變量副本的工作內(nèi)存，線程對(duì)共享變量的操作都在此工作內(nèi)存中進(jìn)行。
-
某個(gè)線程無(wú)法直接訪問(wèn)其他線程中的變量，線程之間的通信需要通過(guò)主內(nèi)存來(lái)實(shí)現(xiàn)。
內(nèi)存模型
線程通信過(guò)程
- 線程A從主內(nèi)存中拷貝共享變量1到工作內(nèi)存中的副本，對(duì)副本的值進(jìn)行修改。
- 線程A刷新修改后的值到主內(nèi)存中。
- 線程B將主內(nèi)存中共享變量1拷貝到工作內(nèi)存中。
并發(fā)編程三個(gè)特征
- 原子性：類(lèi)似于數(shù)據(jù)庫(kù)事務(wù)，要么全部執(zhí)行，要么不執(zhí)行。
如：num++不具有原子性，先取出num的值，再進(jìn)行加1。
而a=1,return a則都具有原子性。 - 可見(jiàn)性：某個(gè)線程對(duì)共享變量做了修改后，其他線程可以立馬感知到該變量的修改。
- 有序性：若在本線程內(nèi)觀察，所有操作都是有序的；若在一個(gè)線程中觀察其他線程，所有的操作都是無(wú)序的。前半句指“線程內(nèi)表現(xiàn)為串行語(yǔ)義”，后半句指“指令重排序”現(xiàn)象和“工作內(nèi)存中主內(nèi)存同步延遲”現(xiàn)象。
Sychronized與Volatile
- Sychronized：可以保證原子性、可見(jiàn)性和有序性。
Sychronized關(guān)鍵字能保證在同一時(shí)刻，只有一個(gè)線程可以獲取鎖執(zhí)行同步代碼，執(zhí)行完之后釋放鎖之前，會(huì)將修改后變量的值刷新的主內(nèi)存中。 - Volatile：可以保證可見(jiàn)性和有序性，不能保證操作的原子性。
被Volatile關(guān)鍵字修飾的變量，在寫(xiě)操作后會(huì)加入一條store指令，強(qiáng)行將共享變量最新的值刷新到主內(nèi)存中。在讀操作前，會(huì)加入一條load指令，強(qiáng)行從主內(nèi)存中讀取共享變量最新的值。
Java鎖機(jī)制
Java中的鎖
- Sychronized
- ReentrantLock
- ReentrantReadWriteLock
這三種鎖是怎么實(shí)現(xiàn)的？什么是AQS？什么是CAS？CAS的ABA問(wèn)題怎么解決？集群環(huán)境下如何實(shí)現(xiàn)同步？有待后續(xù)分曉
Java多線程實(shí)現(xiàn)
創(chuàng)建線程的方式
-
繼承Thread類(lèi)：重寫(xiě)
run()方法，創(chuàng)建線程后調(diào)用start()方法啟動(dòng)。public class ThreadOne extends Thread { private static Integer num = 100; @Override public void run() { synchronized (num) { while (num > 0) { try { Thread.sleep((long) (Math.random() * 100)); } catch (InterruptedException e) { e.printStackTrace(); } System.out.print("線程:" + this.getName() + "執(zhí)行num--操作后，"); num--; System.out.println("num的值為:" + num); } } } public static void main(String[] args) { Thread t1 = new ThreadOne(); Thread t2 = new ThreadOne(); Thread t3 = new ThreadOne(); t1.start(); t2.start(); t3.start(); } } -
實(shí)現(xiàn)Runnable接口：實(shí)現(xiàn)
run()方法，創(chuàng)建線程后調(diào)用start()方法啟動(dòng)。public class ThreadTwo implements Runnable{ private static Integer num = 100; @Override public void run() { synchronized (num) { while (num > 0) { try { Thread.sleep((long) (Math.random() * 100)); } catch (InterruptedException e) { e.printStackTrace(); } System.out.print("線程:" + Thread.currentThread().getName() + "執(zhí)行num--操作后，"); num--; System.out.println("num的值為:" + num); } } } public static void main(String[] args) { Thread t1 = new Thread(new ThreadTwo()); Thread t2 = new Thread(new ThreadTwo()); Thread t3 = new Thread(new ThreadTwo()); t1.start(); t2.start(); t3.start(); } } -
實(shí)現(xiàn)Callable接口：實(shí)現(xiàn)
call()方法，可以創(chuàng)建有返回值的線程。public class ThreadThree implements Callable<Integer> { private static Integer num = 100; @Override public Integer call() throws Exception { synchronized (num) { while (num > 0) { try { Thread.sleep((long) (Math.random() * 100)); } catch (InterruptedException e) { e.printStackTrace(); } num--; } } return num; } public static void main(String[] args) throws InterruptedException, ExecutionException { FutureTask<Integer> task1 = new FutureTask<>(new ThreadThree()); FutureTask<Integer> task2 = new FutureTask<>(new ThreadThree()); FutureTask<Integer> task3 = new FutureTask<>(new ThreadThree()); Thread t1 = new Thread(task1); Thread t2 = new Thread(task2); Thread t3 = new Thread(task3); t1.start(); t2.start(); t3.start(); System.out.println("線程1的返回值:" + task1.get()); System.out.println("線程2的返回值:" + task2.get()); System.out.println("線程3的返回值:" + task3.get()); } }
Executor線程池框架
為什么要用線程池？
- 通過(guò)復(fù)用“池”中的已有線程，減少線程創(chuàng)建和銷(xiāo)毀的開(kāi)銷(xiāo)，提高性能。
- 可以設(shè)置最大并發(fā)線程數(shù)，避免過(guò)多線程競(jìng)爭(zhēng)資源。
Executors創(chuàng)建線程池
-
newFixedThreadPool創(chuàng)建固定線程數(shù)的線程池。若所有線程都處于活動(dòng)狀態(tài)，新提交的任務(wù)會(huì)在隊(duì)列中等待。若某個(gè)線程異常結(jié)束，則線程池會(huì)重新補(bǔ)充一個(gè)新線程。
public static ExecutorService newFixedThreadPool(int nThreads) { return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); } -
newCachedThreadPool創(chuàng)建可緩存的線程池，若線程池中的線程超過(guò)60s未被使用會(huì)被移除。
public static ExecutorService newCachedThreadPool() { return new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>()); } -
newScheduledThreadPool創(chuàng)建定時(shí)線程，可定時(shí)執(zhí)行任務(wù)。
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { return new ScheduledThreadPoolExecutor(corePoolSize); } -
newSingleThreadExecutor創(chuàng)建一個(gè)單線程來(lái)執(zhí)行任務(wù)。
public static ExecutorService newSingleThreadExecutor() { return new FinalizableDelegatedExecutorService (new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>())); }
線程池執(zhí)行任務(wù)
-
使用
execute()方法執(zhí)行Runnable任務(wù)public class RunnableThreadPool { public static void main(String[] args) { ExecutorService eService = Executors.newFixedThreadPool(5); for (int i = 0; i < 10; i++) { eService.execute(new Runnable() { @Override public void run() { System.out.println(Thread.currentThread().getName() + "正在執(zhí)行"); } }); } eService.shutdown(); } } -
使用
submit()方法執(zhí)行Callable任務(wù)public class CallableThreadPool { public static void main(String[] args) { ExecutorService eService = Executors.newFixedThreadPool(5); List<Future<String>> resultList = new ArrayList<Future<String>>(); for (int i = 0; i < 10; i++) { Future<String> future = eService.submit(new Callable<String>() { @Override public String call() throws Exception { return Thread.currentThread().getName() + "已執(zhí)行call方法并成功返回"; } }); resultList.add(future); } eService.shutdown(); for (Future<String> future : resultList) { //等待future返回結(jié)果 while(!future.isDone()); try { System.out.println(future.get()); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } } } }
那么線程池的實(shí)現(xiàn)原理是什么？有待后續(xù)見(jiàn)分曉
