在應(yīng)用中，我們常常需要Thread緩沖池來做一些事以提高程序的效率和并發(fā)性。本文演示了如何利用Queue這種數(shù)據(jù)結(jié)構(gòu)實(shí)現(xiàn)一個(gè)簡單的Thread緩沖池。

一個(gè)Thread緩沖池可以設(shè)計(jì)成以下這樣：緩沖池由幾個(gè)工作Thread和一個(gè)Queue組成，Client負(fù)責(zé)把任務(wù)放到Queue里面（put方法），而工作Thread就依次取出這些任務(wù)并執(zhí)行它們（get方法）。

Queue的一個(gè)經(jīng)典實(shí)現(xiàn)是使用一個(gè)循環(huán)數(shù)組（這個(gè)實(shí)現(xiàn)在很多數(shù)據(jù)結(jié)構(gòu)的書上都有介紹），如一個(gè)大小為size的數(shù)組，這個(gè)循環(huán)數(shù)組可以被想象成首尾相連的一個(gè)環(huán)。oldest指向Queue中最老的數(shù)據(jù)所在的位置，next指向下一個(gè)可以放新數(shù)據(jù)的位置。

放入一個(gè)新數(shù)據(jù)到next的位置后，需要更新next：next = (next + 1) % size;

從oldest位置取出一個(gè)數(shù)據(jù)后，需要更新oldest：oldest = (oldest + 1) % size;

當(dāng)oldest == next的時(shí)候，Queue為空，

當(dāng)(next + 1) % size == oldest的時(shí)候，Queue為滿。

（注意：為了區(qū)分Queue為空和為滿的情況，實(shí)際上Queue里面最多能放size-1個(gè)數(shù)據(jù)。）

因?yàn)檫@個(gè)Queue會(huì)同時(shí)被多個(gè)線程訪問，需要考慮在這種情況下Queue如何工作。首先，Queue需要是線程安全的，可以用Java里的synchronized關(guān)鍵字來確保同時(shí)只有一個(gè)Thread在訪問Queue.

我們還可以注意到當(dāng)Queue為空的時(shí)候，get操作是無法進(jìn)行的；當(dāng)Queue為滿的時(shí)候，put操作又是無法進(jìn)行的。在多線程訪問遇到這種情況時(shí)，一般希望執(zhí)行操作的線程可以等待（block）直到該操作可以進(jìn)行下去。比如，但一個(gè)Thread在一個(gè)空Queue上執(zhí)行g(shù)et方法的時(shí)候，這個(gè) Thread應(yīng)當(dāng)?shù)却?block)，直到另外的Thread執(zhí)行該Queue的put方法后，再繼續(xù)執(zhí)行下去。在Java里面，Object對象的 wait(),notify()方法提供了這樣的功能。

把上面的內(nèi)容結(jié)合起來，就是一個(gè)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是會(huì)有問題的。這是個(gè)很容易犯的錯(cuò)誤。;-)
在以上代碼中使用了ExceptionAdapter這個(gè)類，它的作用是把一個(gè)checked Exception包裝成RuntimeException。詳細(xì)的說明可以參考我的避免在Java中使用Checked Exception一文。

接下來我們需要一個(gè)對象來表現(xiàn)Thread緩沖池所要執(zhí)行的任務(wù)。可以發(fā)現(xiàn)JDK中的Runnable interface非常合適這個(gè)角色。

最后，剩下工作線程的實(shí)現(xiàn)就很簡單了：從SyncQueue里取出一個(gè)Runnable對象并執(zhí)行它。

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;

}

下面是一個(gè)使用這個(gè)Thread緩沖池的例子：

       //構(gòu)造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);

為了使本文中的代碼盡可能簡單，這個(gè)Thread緩沖池的實(shí)現(xiàn)是一個(gè)基本的框架。當(dāng)使用到實(shí)際中時(shí)，一些其他功能也可以在這一基礎(chǔ)上添加，比如異常處理，動(dòng)態(tài)調(diào)整緩沖池大小等等。