在應用中,我們常常需要Thread緩沖池來做一些事以提高程序的效率和并發性。本文演示了如何利用Queue這種數據結構實現一個簡單的Thread緩沖池。

 

一個Thread緩沖池可以設計成以下這樣:緩沖池由幾個工作Thread和一個Queue組成,Client負責把任務放到Queue里面(put方法),而工作Thread就依次取出這些任務并執行它們(get方法)。

 

Queue的一個經典實現是使用一個循環數組(這個實現在很多數據結構的書上都有介紹),如下圖所示:


圖示是一個大小為size的數組,這個循環數組可以被想象成首尾相連的一個環。oldest指向Queue中最老的數據所在的位置,next指向下一個可以放新數據的位置。

放入一個新數據到next的位置后,需要更新nextnext = (next + 1) % size;

oldest位置取出一個數據后,需要更新oldestoldest = (oldest + 1) % size;

oldest == next的時候,Queue為空,

(next + 1) % size == oldest的時候,Queue為滿。

(注意:為了區分Queue為空和為滿的情況,實際上Queue里面最多能放size-1個數據。)

 

因為這個Queue會同時被多個線程訪問,需要考慮在這種情況下Queue如何工作。首先,Queue需要是線程安全的,可以用Java里的synchronized關鍵字來確保同時只有一個Thread在訪問Queue.

 

我們還可以注意到當Queue為空的時候,get操作是無法進行的;當Queue為滿的時候,put操作又是無法進行的。在多線程訪問遇到這種情況時,一般希望執行操作的線程可以等待(block)直到該操作可以進行下去。比如,但一個Thread在一個空Queue上執行get方法的時候,這個Thread應當等待(block),直到另外的Thread執行該Queueput方法后,再繼續執行下去。在Java里面,Object對象的wait(),notify()方法提供了這樣的功能。

 

把上面的內容結合起來,就是一個SyncQueue的類:

public class SyncQueue {

   

    public SyncQueue(int size) {

       _array = new Object[size];

       _size = size;

       _oldest = 0;

       _next = 0;

    }

   

    public synchronized void put(Object o) {

       while (full()) {

           try {

              wait();

           } catch (InterruptedException ex) {

              throw new ExceptionAdapter(ex);

           }

       }

       _array[_next] = o;

       _next = (_next + 1) % _size;

       notify();

    }

   

    public synchronized Object get() {

       while (empty()) {

           try {

              wait();

           } catch (InterruptedException ex) {

              throw new ExceptionAdapter(ex);

           }

       }

       Object ret = _array[_oldest];

       _oldest = (_oldest + 1) % _size;

       notify();

       return ret;

    }

   

    protected boolean empty() {

       return _next == _oldest;

    }

   

    protected boolean full() {

       return (_next + 1) % _size == _oldest;

    }

   

    protected Object [] _array;

    protected int _next;

    protected int _oldest;

    protected int _size;

}


可以注意一下getput方法中while的使用,如果換成if是會有問題的。這是個很容易犯的錯誤。;-)
在以上代碼中使用了ExceptionAdapter這個類,它的作用是把一個checked Exception包裝成RuntimeException。詳細的說明可以參考我的避免在Java中使用Checked Exception一文。

 

接下來我們需要一個對象來表現Thread緩沖池所要執行的任務。可以發現JDK中的Runnable interface非常合適這個角色。

 

最后,剩下工作線程的實現就很簡單了:從SyncQueue里取出一個Runnable對象并執行它。

public class Worker implements Runnable {

   

    public Worker(SyncQueue queue) {

       _queue = queue;

    }

   

    public void run() {

       while (true) {

           Runnable task = (Runnable) _queue.get();

           task.run();

       }

    }

   

    protected SyncQueue _queue = null;

}

 

下面是一個使用這個Thread緩沖池的例子:

       //構造Thread緩沖池

       SyncQueue queue = new SyncQueue(10);

       for (int i = 0; i < 5; i ++) {

           new Thread(new Worker(queue)).start();

       }

      

       //使用Thread緩沖池

       Runnable task = new MyTask();

       queue.put(task);

 

為了使本文中的代碼盡可能簡單,這個Thread緩沖池的實現是一個基本的框架。當使用到實際中時,一些其他功能也可以在這一基礎上添加,比如異常處理,動態調整緩沖池大小等等。


 

作者DaiJiaLin