对 ThreadPool 的理解

虽然从 Java5 开始 JDK 里的 java.util.concurrent 包内建了线程池，你不必自己实现线程池，但理解线程的实现原理对 Java 编程很有用。
当你想把运行在你的程序中的线程控制在一定的数量之内，线程池就显得非常有用。

引用

原理：

用一个阻塞队列里(Blocking Queue)来存储线程池的所有空闲线程。不用为每个任务都创建一个新的线程，可以把任务当参数传到线程池里。只要当线程池有空闲的线程时，这个任务就会被执行。

BlockQueue.java

public class BlockingQueue {

    private List queue = new LinkedList();
    private int limit = 10;

    public BlockingQueue(int limit) {
        this.limit = limit;
    }

    public synchronized Object enqueue(Object item)
            throws InterruptedException {
        while (this.queue.size() == this.limit) {
            System.out.println("Arrive to the pool's max size:"+this.limit);
            wait();
        }
        if (this.queue.size() == 0) {
            notifyAll();
        }

        System.out.println("Add a new thread to the pool.Pool's size before 

adding:"+this.queue.size());
        this.queue.add(item);
        
        return this.queue.get(this.queue.size()-1);
    }

    public synchronized Object dequeue()
            throws InterruptedException {
        while (this.queue.size() == 0) {
            System.out.println("Pool's size is 0.");
            wait();
        }
        if (this.queue.size() == this.limit) {
            notifyAll();
        }

        System.out.println("Remove a thread from the pool.Pool's size before 

removing:"+(this.queue.size()-1));
        return this.queue.remove(0);
    }
    
    public synchronized int size(){
        return this.queue.size();
    }
}

当阻塞队列没达到界限值（最大值与最小值）时，插入与出列正常，没限制。阻塞队列达到最大值时，再想插入一个线程，队列就会停止操作，让插入在等待，同时唤醒出列操作。相反，当全部出列后，再想出列时，出列操作就会停止，在等待，同时唤醒插入操作。


线程池通常用在多线程服务端的，下边是实现线程池的其他类，其中：

/**
 * Implements the thread pool
 * @author winxp
 */
public class ThreadPool {

    private BlockingQueue taskQueue = null;
    private List<TaskThread> threads = new ArrayList<TaskThread>();
    private boolean isStopped = false;

    /**
     * Create a new ThreadPool
     * @param noOfThreads Initiate num of threads in the thread pool线程池的初始化线程数
     * @param maxNoOfTasks Max num of threads in the thread pool线程池的最大线程数
     */
    public ThreadPool(int noOfThreads, int maxNoOfTasks) {
        taskQueue = new BlockingQueue(maxNoOfTasks);

        for (int i = 0; i < noOfThreads; i++) {
            threads.add(new TaskThread(taskQueue));
        }
        for (TaskThread thread : threads) {
            thread.start();
        }
    }
    
    /**
     *
     * @param task
     */
    public synchronized void execute(Runnable task){
        if(this.isStopped)throw 
                new IllegalStateException("ThreadPool is stopped");
        try {
            this.taskQueue.enqueue(task);
        } catch (InterruptedException ex) {
           ex.printStackTrace();
        }
    }

    /**
     *
     */
    public synchronized void stop() {
        this.isStopped = true;
        for (TaskThread thread : threads) {
            thread.interrupt();
        }
    }
}

/**
 * Simulate that dequeue from the BlockingQueue, and execute the dequeuing task
 */
public class TaskThread extends Thread {

    private BlockingQueue taskQueue = null;
    private boolean isStopped = false;

    public TaskThread(BlockingQueue queue) {
        taskQueue = queue;
    }

    @Override
    public void run() {
        while (!isStopped()) {
            try {
                Runnable runnable = (Runnable) taskQueue.dequeue();
                runnable.run();
            } catch (Exception e) {
                //log or otherwise report exception,
                //but keep pool thread alive.
            }
        }
    }
    
    @Override
    public synchronized void interrupt() {
        isStopped = true;
        super.interrupt();//break pool thread out of dequeue() call.
    }

    public synchronized boolean isStopped() {
        return isStopped;
    }
}

测试类：

/**
 * Testing class
 */
class TestThread extends Thread{
    private int num;
    public TestThread(int num){
        this.num = num;
    }
    
    @Override
    public void run(){
        System.out.println("TestThread index: "+num);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException ex) {
        }
    }

    /**
     *
     * @param args
     */
    public static void main(String[] args){
        ThreadPool tp = new ThreadPool(3,10);
        for(int i = 0;i<15;i++){
            TestThread test = new TestThread(i);
            tp.execute(test);
        }
    }
}

评论

4 楼 天下无贼 2012-04-03  

Sorry，看错了。是run方法（）

3 楼 天下无贼 2012-04-03  

ThreadPool 这个类的68,69行似乎有问题。Ruunable接口应该没有start()方法吧？

2 楼 fhyfufangyu 2011-05-23  

truth315 写道

 while (this.queue.size() == this.limit) {   
           System.out.println("Arrive to the pool's max size:"+this.limit);   
            wait();   
       }

为什么不能把WHILE改为IF，然后把唤醒的notifyAll()改为notify();

为什么用while可以参考jdk 文档中“虚假唤醒”的描述：

在没有被通知、中断或超时的情况下，线程还可以唤醒一个所谓的虚假唤醒 (spurious wakeup)。虽然这种情况在实践中很少发生，但是应用程序必须通过以下方式防止其发生，即对应该导致该线程被提醒的条件进行测试，如果不满足该条件，则继续等待。换句话说，等待应总是发生在循环中，如下面的示例：

synchronized (obj) {
while (<condition does not hold>)
obj.wait(timeout);
... // Perform action appropriate to condition
     }

1 楼 truth315 2010-05-12  

 while (this.queue.size() == this.limit) {   
           System.out.println("Arrive to the pool's max size:"+this.limit);   
            wait();   
       }

为什么不能把WHILE改为IF，然后把唤醒的notifyAll()改为notify();