《Java高并發程序設計》總結--2.Java并行程序基礎

2019-11-06 07:26:28

字體：大中小

來源：轉載

供稿：網友

2.1 基本概念1）進程進程（PRocess）是計算機中的程序關于某數據集合上的一次運行活動，是系統進行資源分配和調度的基本單位，是操作系統結構的基礎。程序是指令、數據及其組織形式的描述，進程是程序的實體。2）線程線程是輕量級進程，是程序執行的最小單位。使用多線程而不是多進程去進行并發程序設計，是因為線程間的切換或調度的成本遠小于進程。3）線程的狀態：NEW狀態表示剛剛創建的線程，這種線程還沒開始執行。等到線程的start()方法調用時，才表示開始執行。當線程執行時，處于RUNNABLE狀態，表示線程所需的一切資源都已經準備好了。如果線程在執行過程中遇到了synchronized同步塊，就會進入BLOCKED阻塞狀態。這事線程就會暫停執行，直到獲得請求的鎖。WAITING和TIMED_WAITING都表示等待狀態，它們的區別是WAITING會進入一個無時間限制的等待，TIME_WATING會進入一個有時限的等待。一旦等到了期望的事件，線程會再次執行，進入RUNNABLE狀態。當線程執行完畢后，則進入TERMINATED狀態，表示結束。2.2 初始線程：線程的基本操作1）新建線程java提供了線程類Thread來創建多線程的程序。其實，創建線程與創建普通的類的對象的操作是一樣的，而線程就是Thread類或其子類的實例對象。每個Thread對象描述了一個單獨的線程。要產生一個線程，有兩種方法：◆ 需要從Java.lang.Thread類派生一個新的線程類，重載它的run()方法； ◆ 實現Runnalbe接口，重載Runnalbe接口中的run()方法。2）終止線程Thread.stop()方法在結束線程時，會直接終止線程，并且會立即釋放這個線程所持有的鎖。而這些鎖恰恰是用來維持對象一致性的。如果此時，寫線程寫入數據正寫到一半，并強行終止，那么對象就會被破壞，同時，由于鎖已經被釋放，另外一個等待該鎖的讀線程就順理成章的讀到了這個不一致的對象。

這個過程可以用一下代碼模擬package%20cn.guet.parallel;public%20class%20StopThreadUnsafe%20{public%20static%20User%20u%20=%20new%20User();public%20static%20class%20User%20{private%20int%20id;private%20String%20name;public%20User()%20{id%20=%200;name%20=%20"0";}public%20int%20getId()%20{return%20id;}public%20void%20setId(int%20id)%20{this.id%20=%20id;}public%20String%20getName()%20{return%20name;}public%20void%20setName(String%20name)%20{this.name%20=%20name;}@Overridepublic%20String%20toString()%20{//%20TODO%20Auto-generated%20method%20stubreturn%20"User%20[id="%20+%20id%20+%20",name="%20+%20name%20+%20"]";}}public%20static%20class%20ChangeObjectThread%20extends%20Thread%20{@Overridepublic%20void%20run()%20{while(true)%20{synchronized%20(u)%20{int%20v%20=%20(int)(System.currentTimeMillis()/1000);u.setId(v);try%20{Thread.sleep(100);}%20catch%20(Exception%20e)%20{e.printStackTrace();}u.setName(String.valueOf(v));}Thread.yield();}}}public%20static%20class%20ReadObjectThread%20extends%20Thread%20{@Overridepublic%20void%20run()%20{while%20(true)%20{synchronized%20(u)%20{if(u.getId()%20!=%20Integer.parseInt(u.getName()))%20{System.out.println(u.toString());}};%20}}}public%20static%20void%20main(String[]%20args)%20throws%20Exception%20{new%20ReadObjectThread().start();while%20(true)%20{Thread%20t%20=%20new%20ChangeObjectThread();t.start();Thread.sleep(150);t.stop();}}}如果需要停止一個線程，只是需要自行決定線程何時退出就可以。用上述例子說明，只需將ChangeObjectTread線程增加一個stopMe()方法即可。public%20static%20class%20ChangeObjectThread%20extends%20Thread%20{volatile%20boolean%20stopme%20=%20false;public%20void%20stopMe()%20{stopme%20=%20true;}@Overridepublic%20void%20run()%20{while(true)%20{if(stopme)%20{System.out.println("exit%20by%20stop%20me");break;}synchronized%20(u)%20{int%20v%20=%20(int)(System.currentTimeMillis()/1000);u.setId(v);try%20{Thread.sleep(100);}%20catch%20(Exception%20e)%20{e.printStackTrace();}u.setName(String.valueOf(v));}Thread.yield();}}}3）線程中斷Thread.interrupt()方法是一個實例方法。它通知目標線程中斷，也就是設置中斷標志位。中斷標志位表示當前線程已經被中斷。Thread.isInterrupted()方法也是實例方法，它判斷當前線程是否有被中斷(通過檢查中斷標志位)。最后的靜態方法Thread.interrupted()也是用來判斷當前線程的中斷狀態，但同時會清除當前線程的中斷標志位狀態。public%20static%20void%20main(String[]%20args)%20throws%20Exception%20{new%20ReadObjectThread().start();Thread%20t%20=%20new%20Thread()%20{@Overridepublic%20void%20run()%20{while%20(true)%20{if(Thread.currentThread().isInterrupted())%20{System.out.println("Interruted!");}try%20{Thread.sleep(2000);}%20catch%20(InterruptedException%20e)%20{System.out.println("Interruted%20When%20Sleep");Thread.currentThread().interrupt();}Thread.yield();}}};t.start();Thread.sleep(2000);t.interrupt();}4）等待和通知當一個對象實例上調用wait()方法后，當前線程就會在這個對象上等待。比如，線程A中，調用了obj.wait()方法，那么線程A就會停止繼續執行，而轉為等待狀態。線程A一直等到其他線程調用了obj.notify()方法為止。這時，obj對象就儼然成為多個線程之間的有效通信手段。一個簡單地使用wait()和notify()的案例：public class SimpleWN {final static Object object = new Object();public static class T1 extends Thread {@Overridepublic void run() {synchronized (object) {System.out.println(System.currentTimeMillis() + "T1 start!");try {System.out.println(System.currentTimeMillis() + "T1 wait for object");object.wait();} catch (Exception e) {e.printStackTrace();}System.out.println(System.currentTimeMillis() + "T1 end!");}}}public static class T2 extends Thread {@Overridepublic void run() {synchronized (object) {System.out.println(System.currentTimeMillis() + "T2 start! notify one thread");object.notify();System.out.println(System.currentTimeMillis() + "T2 end!");try {Thread.sleep(2000);} catch (InterruptedException e) {// TODO: handle exception}}}}public static void main(String[] args) {Thread t1 = new T1();Thread t2 = new T2();t1.start();t2.start();}}5）掛起和繼續執行線程線程掛起(suspend)和繼續執行(resume)是一對相反的操作，被掛起的線程，必須要等到resume()操作后，才能繼續執行。并不推薦使用suspend()去掛起線程，因為suspend()在導致線程暫停的同時，并不會去釋放任何資源。此時，其他線程想要訪問被它暫用的鎖時，都會被牽連，導致無法正常繼續運行。為理解suspend()的問題，演示程序如下：public class BadSuspend {public static Object u = new Object();static ChangeObjectThread t1 = new ChangeObjectThread("t1");static ChangeObjectThread t2 = new ChangeObjectThread("t2");public static class ChangeObjectThread extends Thread {public ChangeObjectThread(String name) {super.setName(name);}@Overridepublic void run() {synchronized (u) {System.out.println("in " + getName());Thread.currentThread().suspend();}}}public static void main(String[] args) throws InterruptedException {t1.start();Thread.sleep(100);t2.start();t1.resume();t2.resume();t1.join();t2.join();}}改進后的代碼如下：public class GoodSuspend {public static Object u = new Object();public static class ChangeObjectThread extends Thread {volatile boolean suspendme = false;public void suspendMe() {suspendme = true;}public void resumeMe() {suspendme = false;synchronized (this) {notify();}}@Overridepublic void run() {while (true) {synchronized (this) {while (suspendme) {try {wait();} catch (InterruptedException e) {e.printStackTrace();}}synchronized (u) {System.out.println("in ChangeObjectThread");}Thread.yield();}}}public static class ReadObjectThread extends Thread {@Overridepublic void run() {while(true) {synchronized (u) {System.out.println("in ReadObjectThread");}Thread.yield();}}}public static void main(String[] args) throws InterruptedException {ChangeObjectThread t1 = new ChangeObjectThread();ReadObjectThread t2 = new ReadObjectThread();t1.start();t2.start();Thread.sleep(1000);t1.suspendMe();System.out.println("suspend t1 2 second");Thread.sleep(2000);System.out.println("resume t1");t1.resumeMe();}}}6）等待線程結束(join)和謙讓(yield)第一個join()方法表示無限等待，它會一直阻塞當前線程，直到目標線程執行完畢。第二個方法給出了一個最大等待時間，如果超過給定時間目標線程還在執行，當前線程也會因為“等不及了”，而繼續往下執行。這里提供一個簡單的join實例：public class JoinMain {public volatile static int i = 0;public static class AddThread extends Thread {@Overridepublic void run() {for (i = 0; i < 10000000; i++) {}}}public static void main(String[] args) throws InterruptedException {AddThread at = new AddThread();at.start();at.join();System.out.println(i);}}主函數中，如果不使用join()等待AddThread，那么得到i很可能是0或者一個非常小的數字。因為AddThread還沒開始執行，i的值就已經被輸出了。但在使用join()方法后，表示主線程愿意等待AddThread執行完畢，跟著AddTread一起往前走，故在join()返回時，AddThread已經執行完成，故i總是10000000。2.3 volatile與Java內存模型(JMM)volatile的定義如下： java編程語言允許線程訪問共享變量，為了確保共享變量能被準確和一致的更新，線程應該確保通過排他鎖單獨獲得這個變量。volatile對于保證操作的原子性是有非常大的幫助的。但是，volatile并不能代替鎖，它無法保證一些復合操作的原子性，例如，volatile是無法保證i++的原子性操作的：public class PlusTask {static volatile int i = 0;public static class Plus implements Runnable {public void run() {for (int k = 0; k < 10000; k++) {i++;}}}public static void main(String[] args) throws InterruptedException {Thread[] threads = new Thread[10];for (int i = 0; i < 10; i++) {threads[i] = new Thread(new Plus());threads[i].start();}for (int i = 0; i < 10; i++) {threads[i].join();}System.out.println(i);}}此外，volatile也能保證數據的可見性和有序性。public class NoVisibility {private static boolean ready;private static int number;private static class ReaderThread extends Thread {@Overridepublic void run() {while(!ready);System.out.println(number);}}public static void main(String[] args) throws InterruptedException {new ReaderThread().start();Thread.sleep(1000);number = 42;ready = true;Thread.sleep(10000);}}在虛擬機的Client模式下，由于JIT并沒有做足夠的優化，在主線程修改ready變量的狀態后，ReaderThread可以發現這個改動，并退出程序。但是在Server模式下，由于系統優化的結果，ReaderThread線程無法“看到”主線程中的修改，導致ReaderThread永遠無法退出。這個問題就是一個典型的可見性問題。2.4 分門別類的管理：線程組線程組的使用如下：public class ThreadGroupName implements Runnable {@Overridepublic void run() {String groupAndName = Thread.currentThread().getThreadGroup().getName()+ "-" + Thread.currentThread().getName();while(true) {System.out.println("I am " + groupAndName);try {Thread.sleep(3000);} catch (InterruptedException e) {e.printStackTrace();}}}public static void main(String[] args) {ThreadGroup tg = new ThreadGroup("PrintGroup");Thread t1 = new Thread(tg,new ThreadGroupName(),"T1");Thread t2 = new Thread(tg,new ThreadGroupName(),"T2");t1.start();t2.start();System.out.println(tg.activeCount());tg.list();}}2.5 守護線程(Daemon)只要當前JVM實例中尚存在任何一個非守護線程沒有結束，守護線程就全部工作；只有當最后一個非守護線程結束時，守護線程隨著JVM一同結束工作。Daemon的作用是為其他線程的運行提供便利服務，守護線程最典型的應用就是 GC (垃圾回收器)，它就是一個很稱職的守護者。這里有幾點需要注意： (1) thread.setDaemon(true)必須在thread.start()之前設置，否則會跑出一個IllegalThreadStateException異常。你不能把正在運行的常規線程設置為守護線程。(2) 在Daemon線程中產生的新線程也是Daemon的。 (3) 不要認為所有的應用都可以分配給Daemon來進行服務，比如讀寫操作或者計算邏輯。2.6 線程優先級下面的代碼展示了優先級的作用：public class PriorityDemo {public static class HightPriority extends Thread {static int count = 0;@Overridepublic void run() {while(true) {synchronized (PriorityDemo.class) {count++;if(count > 10000000) {System.out.println("HightPriority is complete");break;}}}}}public static class LowPriority extends Thread {static int count = 0;@Overridepublic void run() {while (true) {synchronized (PriorityDemo.class) {count++;if(count > 10000000) {System.out.println("LowPriority is complete");break;}}}}}public static void main(String[] args) {Thread high = new HightPriority();LowPriority low = new LowPriority();high.setPriority(Thread.MAX_PRIORITY);low.setPriority(Thread.MIN_PRIORITY);low.start();high.start();}}2.7 線程安全的概念與synchronized下面的代碼演示了一個計數器，兩個線程同時對i進行累加操作，各執行10000000次。在很多時候，i的最終值會小于20000000。這就是因為兩個線程同時對i進行寫入時，其中一個線程的結果會覆蓋另一個。public class AccountingVol implements Runnable {static AccountingVol instance = new AccountingVol();static volatile int i =0;public static void increase() {i++;}@Overridepublic void run() {for(int j=0; j<10000000; j++) {increase();}}public static void main(String[] args) throws InterruptedException {Thread t1 = new Thread(instance);Thread t2 = new Thread(instance);t1.start();t2.start();t1.join();t2.join();System.out.println(i);}}要從根本上解決這個問題，我們就必須保證對個線程對i進行操作時完全同步。也就是說，當線程A在寫入時，線程B不僅不能寫，同時也不能讀。關鍵字synchronized可以有多種用法。1）指定加鎖對象：對給定對象加鎖，進入同步代碼前要獲得給定對象的鎖。2）直接作用于實例方法：相當于對當前實例加鎖，進入同步代碼前要獲得當前實例的鎖。3）直接作用于靜態方法：相當于對當前類加鎖，進入同步代碼前要獲得當前類的鎖。下述代碼中，將synchronized作用于一個給定對象instance，因此，每次當線程進入被synchronized包裹的代碼段，就都會要求請求instance實例的鎖。如果當前有其他線程正持有這把鎖，那么新到的線程就必須等待。這樣，就保證了每次只能有一個線程執行i++操作。public class AccountingSync implements Runnable{static AccountingSync instance = new AccountingSync();static int i = 0;@Overridepublic void run() {for(int j=0; j<10000000; j++) {synchronized (instance) {i++;}}}public static void main(String[] args) throws InterruptedException {Thread t1 = new Thread(instance);Thread t2 = new Thread(instance);t1.start();t2.start();t1.join();t2.join();System.out.println(i);}}2.8 隱蔽的錯誤1）無提示的錯誤案例int v1 = 1073741827;int v2 = 1431655768;int ave = (v1+v2)/2;System.out.println(ave);上述代碼中，視圖計算v1和v2的均值。這是一個典型的溢出問題。v1+v2的結果已經導致了int的溢出。2）并發下的ArrayListArrayList是一個線程不安全的容器。public class ArrayListMultiThread {static ArrayList<Integer> al = new ArrayList<Integer>();public static class AddThread implements Runnable {@Overridepublic void run() {for (int i = 0; i < 1000000; i++) {al.add(i);}}}public static void main(String[] args) throws InterruptedException {Thread t1 = new Thread(new AddThread());Thread t2 = new Thread(new AddThread());t1.start();t2.start();t1.join();t2.join();System.out.println(al.size());}}執行這段代碼，可能出現三種結果。第一，程序正常結束。第二，程序拋出異常：Exception in thread "Thread-0" java.lang.ArrayIndexOutOfBoundsException: 6246這是因為ArrayList在擴容過程中，內部一致性被破壞，但由于沒有鎖的保護，另外一個線程訪問到了不一致的內部狀態，導致出現越界問題。第三，出現了一個非常隱蔽的錯誤比如打印如下值作為結果：1425166這是由于多線程訪問沖突，使得彼此保存容器大小的變量被多線程不正常的訪問，同時兩個線程也同時對ArrayList中的同一個位置進行賦值導致的。3）并發下詭異的HashMapHashMap同樣不是線程安全的。代碼如下：import java.util.HashMap;import java.util.Map;public class HashMapMultiThread {static Map<String, String> map = new HashMap<String, String>();public static class AddThread implements Runnable {int start = 0;public AddThread(int start) {this.start = start;}@Overridepublic void run() {for (int i = start; i < 100000; i+=2) {map.put(Integer.toString(i), Integer.toBinaryString(i));}}}public static void main(String[] args) throws InterruptedException {Thread t1 = new Thread(new HashMapMultiThread.AddThread(0));Thread t2 = new Thread(new HashMapMultiThread.AddThread(1));t1.start();t2.start();t1.join();t2.join();System.out.println(map.size());}}第一，程序正常結束，并且結果也是符合預期的。HashMap的大小為100000。第二，程序正常結束，但結果不符合預期，而是一個小于100000的數字。第三，程序永遠無法結束。前兩種情況，和ArrayList的情況非常相似。而第三種情況，通過查看HashMap.put()方法，可知，由于多線程的沖突，這個鏈表結構已經遭到破壞，鏈表成環了，下述的迭代就等同于一個死循環。但這個死循環的問題在JDK8中已經不存在了。由于JDK8對HashMap的內部做了大規模調整，規避了這個問題。但即使這樣，貿然在多線程環境下使用HashMap依然會導致內部數據不一致。最簡單的解決方案就是使用ConcurrentHashMap代替HashMap。for (Entry<K,V> e = table[i]; e != null; e = e.next) {Object k;if (e.hash == hash && ((k = e.key) == key || key.equals(k))) {V oldValue = e.value;e.value = value;e.record access(this);return oldValue;}}4）錯誤的加鎖假設我們需要一個計數器，這個計數器會被多個線程同時訪問。為了確保數據正確性，我們需要對計數器加鎖。代碼如下:public class BadLockInteger implements Runnable {public static Integer i = 0;static BadLockInteger instance = new BadLockInteger();@Overridepublic void run() {for (int j = 0; j < 10000000; j++) {synchronized (i) {i++;}}}public static void main(String[] args) throws InterruptedException {Thread t1 = new Thread(instance);Thread t2 = new Thread(instance);t1.start();t2.start();t1.join();t2.join();System.out.println(i);}}結果我們得到了一個比20000000小很多的數字。要解釋這個問題，得從Integer說起。在Java中，Integer屬于不變對象。也就是說對象一旦被創建，就不可能被修改。i++在真實執行時變成了：i = Integer.valueOf(i.intValue()+1)。進一步查看 Integer.valueOf()：public static Integer valueOf(int i) {assert IntegerCache.high >= 127;if (i >= IntegerCache.low && i <= IntegerCache.high)return IntegerCache.cache[i + (-IntegerCache.low)];return new Integer(i);}Integer.valueOf()實際上是一個工廠方法，它會傾向于返回一個代表指定數值的Integer實例。因此，i++的本質是，創建一個新的Integer對象，并將它的引用賦值給i。如此一來，我們就明白問題所在，由于在多個線程間，并不一定能夠看到同一個對象(因為i對象一直在變)，因此，兩個線程每次加鎖可能都加在了不同的對象實例上，從而導致對臨界區代碼控制出現問題。注：本篇博客內容摘自《Java高并發程序設計》

上一篇：Java的語句塊

下一篇：摩爾的Java學習筆記1.7