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

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

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

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

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

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

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

8、his鎖this.count = count; /在此釋放this鎖紅色部分表示需要同步的代碼段，該區(qū)域為 危險區(qū)域”，如果兩個以上的線程同時執(zhí)行，會引發(fā)沖突， 因此， 要更改SharedResource的內(nèi)部狀態(tài)， 必須先獲得SharedResource實例的鎖。退出synchronized塊時，線程擁有的鎖自動釋放，于是，別的線程又可以獲取該鎖了。為了提高性能，不一定要鎖定this,例如，SharedResource有兩個獨立變化的變量：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; 若同步整個方法，貝U setA()的時候無法setB(), setB()時無法setA()。為了提高性能，可以使用不同對象的鎖：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)工作。例如，瀏覽器的一個顯示圖片的線程displayThread想要執(zhí)行顯示圖片的任務(wù)，必須等待下載線程downloadThread將該圖片下載完畢。如果圖片還沒有下載完，displayThread可以暫停，當downloadThread完成了任務(wù)后，再通知displayThread圖片準備完畢，可以顯示了，這時，displayThr

11、ead繼續(xù)執(zhí)行。以上邏輯簡單的說就是：如果條件不滿足，則等待。當條件滿足時，等待該條件的線程將被喚醒。在Java中，這個機制的實現(xiàn)依賴于wait/notify。等待機制與鎖機制是密切關(guān)聯(lián)的。例如：synchronized(obj) (while(!condition) (obj.wait();obj.doSomething();當線程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。#當obj.wait()方法返回后，線程A需要再次獲得obj鎖，才能繼續(xù)執(zhí)行。#如果A1,A2,A3都在obj.wait()，貝U B調(diào)用obj.notify()只能喚醒A1,A2,A3中的一個(具體 哪一個由JVM決定)。# obj.notifyAll()則能全

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

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

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

16、請求隊列：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();藍色部分就是服務(wù)器線程的等待條件，而客戶端線程在放入了一個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ù)器線程在紅色部分可能會阻塞，也就是說，Queue.getRequest是一個阻塞方法。這和java標準庫的許多IO方法類似。最后，寫一個Main來啟動他們：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)試，打印出所有線程的當前狀態(tài)。最后，Main負責完成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個Task,然后動態(tài)添加了一些服務(wù)線程，并定期打印線程狀態(tài)，運 行結(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.仔細觀察：一開始只有兩個服務(wù)器線程，因此線程狀態(tài)都是忙，后來線程數(shù)增多，6個線程中的兩個狀態(tài)變成idle,說明處于wait()狀態(tài)。思考：本例的線程調(diào)度算法其實根本沒有，因 為 這 個 應 用 是 圍 繞TaskQueue設(shè)計的，不是以Thread Pool為中心設(shè)計的。因此，Task調(diào)度取決于TaskQueue的getTask()方法，你可以改進這個方法，例如使用優(yōu)先隊列，使優(yōu)先級高的任