一個簡單的Thread緩沖池的實作

在應用中，我們常常需要 Thread 緩沖池來做一些事以提高程式的效率和并發性。本文示範了如何利用 Queue 這種資料結構實作一個簡單的 Thread 緩沖池。

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

Queue 的一個經典實作是使用一個循環數組（這個實作在很多資料結構的書上都有介紹），如一個大小為 size 的數組，這個循環數組可以被想象成首尾相連的一個環。 oldest 指向 Queue 中最老的資料所在的位置， next 指向下一個可以放新資料的位置。

放入一個新資料到 next 的位置後，需要更新 next ： next = (next + 1) % size;

從 oldest 位置取出一個資料後，需要更新 oldest ： oldest = (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 執行該 Queue 的 put 方法後，再繼續執行下去。在 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;

}

可以注意一下 get 和 put 方法中 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 緩沖池的實作是一個基本的架構。當使用到實際中時，一些其他功能也可以在這一基礎上添加，比如異常處理，動态調整緩沖池大小等等。