Java多線程設計模式之線程池模式
前序：
Thread-Per-Message Pattern,是一種對于每個命令或請求，都分配一個線程，由這個線程執行工作。它將“委托消息的一端”和“執行消息的一端”用兩個不同的線程來實現。該線程模式主要包括三個部分：
1，Request參與者(委托人)，也就是消息發送端或者命令請求端
2，Host參與者，接受消息的請求，負責為每個消息分配一個工作線程。
3，Worker參與者，具體執行Request參與者的任務的線程，由Host參與者來啟動。
由于常規調用一個方法后，必須等待該方法完全執行完畢后才能繼續執行下一步操作，而利用線程后，就不必等待具體任務執行完畢，就可以馬上返回繼續執行下一步操作。
背景：
由于在Thread-Per-Message Pattern中對于每一個請求都會生成啟動一個線程，而線程的啟動是很花費時間的工作，所以鑒于此，提出了Worker Thread，重復利用已經啟動的線程。
線程池：
Worker Thread，也稱為工人線程或背景線程，不過一般都稱為線程池。該模式主要在于，事先啟動一定數目的工作線程。當沒有請求工作的時候，所有的工人線程都會等待新的請求過來，一旦有工作到達，就馬上從線程池中喚醒某個線程來執行任務，執行完畢后繼續在線程池中等待任務池的工作請求的到達。
任務池：主要是存儲接受請求的集合，利用它可以緩沖接受到的請求，可以設置大小來表示同時能夠接受最大請求數目。這個任務池主要是供線程池來訪問。
線程池：這個是工作線程所在的集合，可以通過設置它的大小來提供并發處理的工作量。對于線程池的大小，可以事先生成一定數目的線程，根據實際情況來動態增加或者減少線程數目。線程池的大小不是越大越好，線程的切換也會耗時的。
存放池的數據結構，可以用數組也可以利用集合，在集合類中一般使用Vector，這個是線程安全的。
Worker Thread的所有參與者：
1，Client參與者，發送Request的參與者
2，Channel參與者，負責緩存Request的請求，初始化啟動線程，分配工作線程
3，Worker參與者，具體執行Request的工作線程
4，Request參與者
注意：將在Worker線程內部等待任務池非空的方式稱為正向等待。
將在Channel線程提供Worker線程來判斷任務池非空的方式稱為反向等待。
線程池實例1：
利用同步方法來實現，使用數組來作為任務池的存放數據結構。在Channel有緩存請求方法和處理請求方法，利用生成者與消費者模式來處理存儲請求，利用反向等待來判斷任務池的非空狀態。
Channel參與者：
- package whut.threadpool;
- //用到了生產者與消費者模式
- //生成線程池，接受客戶端線程的請求，找到一個工作線程分配該客戶端請求
- public class Channel {
- private static final int MAX_REQUEST = 100;// 并發數目，就是同時可以接受多少個客戶端請求
- //利用數組來存放請求，每次從數組末尾添加請求，從開頭移除請求來處理
- private final Request[] requestQueue;// 存儲接受客戶線程的數目
- private int tail;//下一次存放Request的位置
- private int head;//下一次獲取Request的位置
- private int count;// 當前request數量
- private final WorkerThread[] threadPool;// 存儲線程池中的工作線程
- // 運用數組來存儲
- public Channel(int threads) {
- this.requestQueue = new Request[MAX_REQUEST];
- this.head = 0;
- this.head = 0;
- this.count = 0;
- threadPool = new WorkerThread[threads];
- // 啟動工作線程
- for (int i = 0; i < threadPool.length; i++) {
- threadPool[i] = new WorkerThread("Worker-" + i, this);
- }
- }
- public void startWorkers() {
- for (int i = 0; i < threadPool.length; i++) {
- threadPool[i].start();
- }
- }
- // 接受客戶端請求線程
- public synchronized void putRequest(Request request) {
- // 當Request的數量大于或等于同時接受的數目時候，要等待
- while (count >= requestQueue.length)
- try {
- wait();
- } catch (InterruptedException e) {
- }
- requestQueue[tail] = request;
- tail = (tail + 1) % requestQueue.length;
- count++;
- notifyAll();
- }
- // 處理客戶端請求線程
- public synchronized Request takeRequest() {
- while (count <= 0)
- try {
- wait();
- } catch (InterruptedException e) {
- }
- Request request = requestQueue[head];
- head = (head + 1) % requestQueue.length;
- count--;
- notifyAll();
- return request;
- }
- }
客戶端請求線程：
- package whut.threadpool;
- import java.util.Random;
- //向Channel發送Request請求的
- public class ClientThread extends Thread{
- private final Channel channel;
- private static final Random random=new Random();
- public ClientThread(String name,Channel channel)
- {
- super(name);
- this.channel=channel;
- }
- public void run()
- {
- try{
- for(int i=0;true;i++)
- {
- Request request=new Request(getName(),i);
- channel.putRequest(request);
- Thread.sleep(random.nextInt(1000));
- }
- }catch(InterruptedException e)
- {
- }
- }
- }
工作線程：
- package whut.threadpool;
- //具體工作線程
- public class WorkerThread extends Thread{
- private final Channel channel;
- public WorkerThread(String name,Channel channel)
- {
- super(name);
- this.channel=channel;
- }
- public void run()
- {
- while(true)
- {
- Request request=channel.takeRequest();
- request.execute();
- }
- }
- }
#p#
線程池實例2：
工作線程：
利用同步塊來處理，利用Vector來存儲客戶端請求。在Channel有緩存請求方法和處理請求方法，利用生成者與消費者模式來處理存儲請求，利用正向等待來判斷任務池的非空狀態。
這種實例，可以借鑒到網絡ServerSocket處理用戶請求的模式中，有很好的擴展性與實用性。
利用Vector來存儲，依舊是每次集合的最后一個位置添加請求，從開始位置移除請求來處理。
Channel參與者：
- package whut.threadpool2;
- import java.util.Vector;
- /*
- * 這個主要的作用如下
- * 0,緩沖客戶請求線程(利用生產者與消費者模式)
- * 1,存儲客戶端請求的線程
- * 2,初始化啟動一定數量的線程
- * 3,主動來喚醒處于任務池中wait set的一些線程來執行任務
- */
- public class Channel {
- public final static int THREAD_COUNT=4;
- public static void main(String[] args) {
- //定義兩個集合，一個是存放客戶端請求的，利用Vector，
- //一個是存儲線程的，就是線程池中的線程數目
- //Vector是線程安全的，它實現了Collection和List
- //Vector 類可以實現可增長的對象數組。與數組一樣，
- //它包含可以使用整數索引進行訪問的組件。但Vector 的大小可以根據需要增大或縮小，
- //以適應創建 Vector 后進行添加或移除項的操作。
- //Collection中主要包括了list相關的集合以及set相關的集合,Queue相關的集合
- //注意：Map不是Collection的子類，都是java.util.*下的同級包
- Vector pool=new Vector();
- //工作線程，初始分配一定限額的數目
- WorkerThread[] workers=new WorkerThread[THREAD_COUNT];
- //初始化啟動工作線程
- for(int i=0;i<workers.length;i++)
- {
- workers[i]=new WorkerThread(pool);
- workers[i].start();
- }
- //接受新的任務，并且將其存儲在Vector中
- Object task=new Object();//模擬的任務實體類
- //此處省略具體工作
- //在網絡編程中，這里就是利用ServerSocket來利用ServerSocket.accept接受一個Socket從而喚醒線程
- //當有具體的請求達到
- synchronized(pool)
- {
- pool.add(pool.size(), task);
- pool.notifyAll();//通知所有在pool wait set中等待的線程，喚醒一個線程進行處理
- }
- //注意上面這步驟添加任務池請求，以及通知線程，都可以放在工作線程內部實現
- //只需要定義該方法為static，在方法體用同步塊，且共享的線程池也是static即可
- //下面這步，可以有可以沒有根據實際情況
- //取消等待的線程
- for(int i=0;i<workers.length;i++)
- {
- workers[i].interrupt();
- }
- }
- }
工作線程：
- package whut.threadpool2;
- import java.util.List;
- public class WorkerThread extends Thread {
- private List pool;//任務請求池
- private static int fileCompressed=0;//所有實例共享的
- public WorkerThread(List pool)
- {
- this.pool=pool;
- }
- //利用靜態synchronized來作為整個synchronized類方法，僅能同時一個操作該類的這個方法
- private static synchronized void incrementFilesCompressed()
- {
- fileCompressed++;
- }
- public void run()
- {
- //保證無限循環等待中
- while(true)
- {
- //共享互斥來訪問pool變量
- synchronized(pool)
- {
- //利用多線程設計模式中的
- //Guarded Suspension Pattern,警戒條件為pool不為空，否則無限的等待中
- while(pool.isEmpty())
- {
- try{
- pool.wait();//進入pool的wait set中等待著,釋放了pool的鎖
- }catch(InterruptedException e)
- {
- }
- }
- //當線程被喚醒，需要重新獲取pool的鎖，
- //再次繼續執行synchronized代碼塊中其余的工作
- //當不為空的時候，繼續再判斷是否為空，如果不為空，則跳出循環
- //必須先從任務池中移除一個任務來執行，統一用從末尾添加，從開始處移除
- pool.remove(0);//獲取任務池中的任務，并且要進行轉換
- }
- //下面是線程所要處理的具體工作
- }
- }
- }
