java多線程設(shè)計(jì)模式-WorkerPattern

1、多線程經(jīng)典實(shí)例java語言已經(jīng)內(nèi)置了多線程支持，所有實(shí)現(xiàn)Runnable接口的類都可被啟動(dòng)一個(gè)新線程，新線程會(huì)執(zhí)行該實(shí)例的run()方法，當(dāng)run()方法執(zhí)行完畢后，線程就結(jié)束了。一旦一個(gè)線程執(zhí) 行完畢，這個(gè)實(shí)例就不能再重新啟動(dòng)，只能重新生成一個(gè)新實(shí)例，再啟動(dòng)一個(gè)新線程。Thread類是實(shí)現(xiàn)了Runnable接口的一個(gè)實(shí)例，它代表一個(gè)線程的實(shí)例，并且，啟動(dòng)線程的唯一方法就是通過Thread類的start()實(shí)例方法：Thread t = new Thread();t.start();start()方法是一個(gè)native方法， 它將啟動(dòng)一個(gè)新線程， 并執(zhí)行run()方法。Thread類默認(rèn)的ru

2、n()方法什么也不做就退出了。注意：直接調(diào)用run()方法并不會(huì)啟動(dòng)一個(gè)新線程，它和調(diào)用一個(gè)普通的java方法沒有什么區(qū)別。因此，有兩個(gè)方法可以實(shí)現(xiàn)自己的線程：方法1:自己的類extend Thread，并復(fù)寫run()方法，就可以啟動(dòng)新線程并執(zhí)行自己定義的run()方法。例如：public class MyThread extends Thread public run() System.out.println(MyThread.run();在合適的地方啟動(dòng)線程：new MyThread().start();方法2:如果自己的類已經(jīng)extends另一個(gè)類，就無法直接extendsThread

3、,此時(shí)，必須實(shí)現(xiàn)一個(gè)Runnable接口：public class MyThread extends OtherClass implements Runnable public run() System.out.println(MyThread.run();為了啟動(dòng)MyThread，需要首先實(shí)例化一個(gè)Thread，并傳入自己的MyThread實(shí)例：MyThread myt = new MyThread();Thread t = new Thread(myt); t.start();事實(shí)上，當(dāng)傳入一個(gè)Runnable target參數(shù)給Thread后，Thread的run()方法就會(huì)調(diào)用targ

4、et.run()，參考JDK源代碼：public void run() (if (target != null) (target.run();線程還有一些Name, ThreadGroup, isDaemon等設(shè)置，由于和線程設(shè)計(jì)模式關(guān)聯(lián)很少，這里 就不多說了。由于同一進(jìn)程內(nèi)的多個(gè)線程共享內(nèi)存空間，在Java中，就是共享實(shí)例，當(dāng)多個(gè)線程試圖同 時(shí)修改某個(gè)實(shí)例的內(nèi)容時(shí)，就會(huì)造成沖突，因此，線程必須實(shí)現(xiàn)共享互斥，使多線程同步。最簡(jiǎn)單的同步是將一個(gè)方法標(biāo)記為synchronized,對(duì)同一個(gè)實(shí)例來說，任一時(shí)刻只能有一個(gè)synchronized方法在執(zhí)行。當(dāng)一個(gè)方法正在執(zhí)行某個(gè)synchronized

5、方法時(shí)，其他線程如果想要執(zhí)行這個(gè)實(shí)例的任意一個(gè)synchronized方法，都必須等待當(dāng)前執(zhí)行synchronized方法的線程退出此方法后，才能依次執(zhí)行。但是，非synchronized方法不受影響，不管當(dāng)前有沒有執(zhí)行synchronized方法，非synchronized方法都可以被多個(gè)線程同時(shí)執(zhí)行。此外，必須注意，只有同一實(shí)例的synchronized方法同一時(shí)間只能被一個(gè)線程執(zhí)行，不同實(shí)例的synchronized方法是可以并發(fā)的。例如，class A定義了synchronized方法sync(),則不 同實(shí)例a1.sync()和a2.sync()可以同時(shí)由兩個(gè)線程來執(zhí)行。多線程同步的

6、實(shí)現(xiàn)最終依賴鎖機(jī)制。我們可以想象某一共享資源是一間屋子，每個(gè)人都是一個(gè)線程。當(dāng)A希望進(jìn)入房間時(shí)，他必須獲得門鎖，一旦A獲得門鎖，他進(jìn)去后就立刻將門鎖上，于是B,C,D.就不得不在門外等待，直到A釋放鎖出來后，B,C,D.中的某一人搶到 了該鎖(具體搶法依賴于JVM的實(shí)現(xiàn)，可以先到先得，也可以隨機(jī)挑選)，然后進(jìn)屋又將門 鎖上。這樣，任一時(shí)刻最多有一人在屋內(nèi)(使用共享資源)。Java語言規(guī)范內(nèi)置了對(duì)多線程的支持。對(duì)于Java程序來說，每一個(gè)對(duì)象實(shí)例都有一把鎖”,一旦某個(gè)線程獲得了該鎖， 別的線程如果希望獲得該鎖， 只能等待這個(gè)線程釋放鎖之后。 獲 得鎖的方法只有一個(gè)，就是synchronized關(guān)

7、鍵字。例如：public class SharedResource (private int count = 0;public int getCount() ( return count; public synchronized void setCount(int count) ( this.count = count; 同步方法public synchronized void setCount(int count) ( this.count = count; 事實(shí)上相當(dāng)于：public void setCount(int count) (synchronized(this) ( /在此獲得t

8、his鎖this.count = count; /在此釋放this鎖紅色部分表示需要同步的代碼段，該區(qū)域?yàn)?危險(xiǎn)區(qū)域”，如果兩個(gè)以上的線程同時(shí)執(zhí)行，會(huì)引發(fā)沖突， 因此， 要更改SharedResource的內(nèi)部狀態(tài)， 必須先獲得SharedResource實(shí)例的鎖。退出synchronized塊時(shí)，線程擁有的鎖自動(dòng)釋放，于是，別的線程又可以獲取該鎖了。為了提高性能，不一定要鎖定this,例如，SharedResource有兩個(gè)獨(dú)立變化的變量：public class SharedResouce (private int a = 0;private int b = 0;public synchr

9、onized void setA(int a) ( this.a = a; public synchronized void setB(int b) ( this.b = b; 若同步整個(gè)方法，貝U setA()的時(shí)候無法setB(), setB()時(shí)無法setA()。為了提高性能，可以使用不同對(duì)象的鎖：public class SharedResouce (private int a = 0;private int b = 0;private Object sync_a = new Object();private Object sync_b = new Object();public vo

10、id setA(int a) (synchronized(sync_a) (this.a = a;public synchronized void setB(int b) (synchronized(sync_b) (this.b = b;通常，多線程之間需要協(xié)調(diào)工作。例如，瀏覽器的一個(gè)顯示圖片的線程displayThread想要執(zhí)行顯示圖片的任務(wù)，必須等待下載線程downloadThread將該圖片下載完畢。如果圖片還沒有下載完，displayThread可以暫停，當(dāng)downloadThread完成了任務(wù)后，再通知displayThread圖片準(zhǔn)備完畢，可以顯示了，這時(shí)，displayThr

11、ead繼續(xù)執(zhí)行。以上邏輯簡(jiǎn)單的說就是：如果條件不滿足，則等待。當(dāng)條件滿足時(shí)，等待該條件的線程將被喚醒。在Java中，這個(gè)機(jī)制的實(shí)現(xiàn)依賴于wait/notify。等待機(jī)制與鎖機(jī)制是密切關(guān)聯(lián)的。例如：synchronized(obj) (while(!condition) (obj.wait();obj.doSomething();當(dāng)線程A獲得了obj鎖后，發(fā)現(xiàn)條件condition不滿足，無法繼續(xù)下一處理，于是線程A就wait()。在另一線程B中，如果B更改了某些條件，使得線程A的condition條件滿足了，就可以喚醒線程A：synchronized(obj) (condition = tru

12、e;obj.notify();需要注意的概念是：#調(diào)用obj的wait(), notify()方法前，必須獲得obj鎖，也就是必須寫在synchronized(obj) (.代 碼段內(nèi)。#調(diào)用obj.wait()后，線程A就釋放了obj的鎖，否則線程B無法獲得obj鎖，也就無法在synchronized(obj) (.代碼段內(nèi)喚醒A。#當(dāng)obj.wait()方法返回后，線程A需要再次獲得obj鎖，才能繼續(xù)執(zhí)行。#如果A1,A2,A3都在obj.wait()，貝U B調(diào)用obj.notify()只能喚醒A1,A2,A3中的一個(gè)(具體 哪一個(gè)由JVM決定)。# obj.notifyAll()則能全

13、部喚醒A1,A2,A3 ,但是要繼續(xù)執(zhí)行obj.wait()的下一條語句，必須獲 得obj鎖，因此，A1,A2,A3只有一個(gè)有機(jī)會(huì)獲得鎖繼續(xù)執(zhí)行，例如A1 ,其余的需要等待A1釋放obj鎖之后才能繼續(xù)執(zhí)行。#當(dāng)B調(diào)用obj.notify/notifyAll的時(shí)候，B正持有obj鎖，因此，A1,A2,A3雖被喚醒， 但是 仍無法獲得obj鎖。 直到B退出synchronized塊， 釋放obj鎖后，A1,A2,A3中的一個(gè)才有機(jī) 會(huì)獲得鎖繼續(xù)執(zhí)行。前面講了wait/notify機(jī)制，Thread還有一個(gè)sleep()靜態(tài)方法，它也能使線程暫停一段時(shí)間。sleep與wait的不同點(diǎn)是：sleep并

14、不釋放鎖，并且sleep的暫停和wait暫停是不一樣的。obj.wait會(huì)使線程進(jìn)入obj對(duì)象的等待集合中并等待喚醒。但是wait()和sleep()都可以通過interrupt()方法打斷線程的暫停狀態(tài)， 從而使線程立刻拋出InterruptedException。如果線程A希望立即結(jié)束線程B,則可以對(duì)線程B對(duì)應(yīng)的Thread實(shí)例調(diào)用interrupt方法。 如果此刻線程B正在wait/sleep/join，則線程B會(huì)立刻拋 出InterruptedException ,在catch() 中直接return即可安全地結(jié)束線程。需要注意的是，InterruptedException是線程自己從

15、內(nèi)部拋出的，并不是interrupt()方法拋出的。對(duì)某一線程調(diào)用interrupt()時(shí)，如果該線程正在執(zhí)行普通的代碼，那么該線程根本就不 會(huì)拋出InterruptedException。但是，一旦該線程進(jìn)入到wait()/sleep()/join()后，就會(huì)立刻拋出InterruptedException。GuardedSuspention模式主要思想是:當(dāng)條件不滿足時(shí)，線程等待，直到條件滿足時(shí)，等待該條件的線程被喚醒。我們?cè)O(shè)計(jì)一個(gè)客戶端線程和一個(gè)服務(wù)器線程，客戶端線程不斷發(fā)送請(qǐng)求給服務(wù)器線程，服務(wù)器線程不斷處理請(qǐng)求。當(dāng)請(qǐng)求隊(duì)列為空時(shí)，服務(wù)器線程就必須等待，直到客戶端發(fā)送了請(qǐng)求。先定義一個(gè)

16、請(qǐng)求隊(duì)列：Queuepackage com.crackj2ee.thread;import java.util.*;public class Queue (private List queue = new LinkedList();public synchronized Request getRequest() (while(queue.size()=0) (try (this.wait();catch(InterruptedException ie) (return null;return (Request)queue.remove(0);public synchronized void p

17、utRequest(Request request) (queue.add(request);this.notifyAll();藍(lán)色部分就是服務(wù)器線程的等待條件，而客戶端線程在放入了一個(gè)request后，就使服務(wù)器線程等待條件滿足，于是喚醒服務(wù)器線程?？蛻舳司€程：ClientThreadpackage com.crackj2ee.thread;public class ClientThread extends Thread (private Queue queue;private String clientName;public ClientThread(Queue queue, String

18、 clientName) (this.queue = queue;this.clientName = clientName;public String toString() (return ClientThread- + clientName + ;public void run() (for(int i=0; i100; i+) (Request request = new Request( + (long)(Math.random()*10000);System.out.println(this + send request: + request);queue.putRequest(req

19、uest);try Thread.sleep(long)(Math.random() * 10000 + 1000);catch(InterruptedException ie) System.out.println(this + shutdown.);服務(wù)器線程：ServerThreadpackage com.crackj2ee.thread;public class ServerThread extends Thread private boolean stop = false;private Queue queue;public ServerThread(Queue queue) thi

20、s.queue = queue;public void shutdown() stop = true;errupt();try this.join();catch(InterruptedException ie) public void run() while(!stop) Request request = queue.getRequest();System.out.println(ServerThread handle request: + request);try Thread.sleep(2000);catch(InterruptedException ie) Syst

21、em.out.println(ServerThread shutdown.);服務(wù)器線程在紅色部分可能會(huì)阻塞，也就是說，Queue.getRequest是一個(gè)阻塞方法。這和java標(biāo)準(zhǔn)庫的許多IO方法類似。最后，寫一個(gè)Main來啟動(dòng)他們：package com.crackj2ee.thread;public class Main public static void main(String口args) Queue queue = new Queue();ServerThread server = new ServerThread(queue);server.start();ClientThre

22、ad口clients = new ClientThread5;for(int i=0; i0) WorkerThread t = (WorkerThread)threads.remove(0); t.shutdown();public synchronized void currentStatus() System.out.println(- );System.out.println(Thread count = + threads.size();Iterator it = threads.iterator();while(it.hasNext() WorkerThread t = (Work

23、erThread)it.next();System.out.println(t.getName() + : + (t.isIdle() ? idle : busy); System.out.println(- );currentStatus()方法是為了方便調(diào)試，打印出所有線程的當(dāng)前狀態(tài)。最后，Main負(fù)責(zé)完成main()方法：package com.crackj2ee.thread;public class Main public static void main(String args) TaskQueue queue = new TaskQueue();ThreadPoolpool=ne

24、wThreadPool(queue);for(inti=0;i10;i+)queue.putTask(newCalculateTask();queue.putTask(newTimerTask();pool.addWorkerThread();pool.addWorkerThread();doSleep(8000);pool.currentStatus();pool.addWorkerThread();pool.addWorkerThread();pool.addWorkerThread();pool.addWorkerThread();pool.addWorkerThread();doSle

25、ep(5000); pool.currentStatus(); private static void doSleep(long ms) tryThread.sleep(ms);catch(InterruptedException ie) (main()一開始放入了20個(gè)Task,然后動(dòng)態(tài)添加了一些服務(wù)線程，并定期打印線程狀態(tài)，運(yùn) 行結(jié)果如下：worker-0 start.CalculateTask 0 start.worker-1 start.TimerTask 0 start.TimerTask 0 done.CalculateTask 1 start.CalculateTask 0 do

26、ne.TimerTask 1 start.CalculateTask 1 done.CalculateTask 2 start.TimerTask 1 done.TimerTask 2 start.TimerTask 2 done.CalculateTask 3 start.Thread count = 2worker-0: busyworker-1: busy CalculateTask 2 done. TimerTask 3 start. worker-2 start.CalculateTask 4 start. worker-3 start.TimerTask 4 start. work

27、er-4 start.CalculateTask 5 start. worker-5 start. TimerTask 5 start. worker-6 start.CalculateTask 6 start. CalculateTask 3 done. TimerTask 6 start.TimerTask 3 done. CalculateTask 7 start. TimerTask 4 done.TimerTask 7 start.TimerTask 5 done.CalculateTask 8 start.CalculateTask 4 done. TimerTask 8 star

28、t.CalculateTask 5 done.CalculateTask 9 start. CalculateTask 6 done. TimerTask 9 start. TimerTask6 done.TimerTask 7 done.Thread count = 7worker-0: idleworker-1: busyworker-2: busyworker-3: idleworker-4: busyworker-5: busyworker-6: busy CalculateTask 7 done.CalculateTask 8 done.TimerTask 8 done.TimerTask 9 done.CalculateTask 9 done.仔細(xì)觀察：一開始只有兩個(gè)服務(wù)器線程，因此線程狀態(tài)都是忙，后來線程數(shù)增多，6個(gè)線程中的兩個(gè)狀態(tài)變成idle,說明處于wait()狀態(tài)。思考：本例的線程調(diào)度算法其實(shí)根本沒有，因 為 這 個(gè) 應(yīng) 用 是 圍 繞TaskQueue設(shè)計(jì)的，不是以Thread Pool為中心設(shè)計(jì)的。因此，Task調(diào)度取決于TaskQueue的getTask()方法，你可以改進(jìn)這個(gè)方法，例如使用優(yōu)先隊(duì)列，使優(yōu)先級(jí)高的任