线程池的实现(转)

线程池功能

应用程序可以有多个线程，这些线程在休眠状态中需要耗费大量时间来等待事件发生。其他线程可能进入睡眠状态，并且仅定期被唤醒以轮循更改或更新状态信息，然后再次进入休眠状态。为了简化对这些线程的管理，.NET框架为每个进程提供了一个线程池，一个线程池有若干个等待操作状态，当一个等待操作完成时，线程池中的辅助线程会执行回调函数。线程池中的线程由系统管理，程序员不需要费力于线程管理，可以集中精力处理应用程序任务。

　　线程池是一种多线程处理形式，处理过程中将任务添加到队列，然后在创建线程后自动启动这些任务。线程池线程都是后台线程.每个线程都使用默认的堆栈大小,以默认的优先级运行,并处于多线程单元中.如果某个线程在托管代码中空闲(如正在等待某个事件),则线程池将插入另一个辅助线程来使所有处理器保持繁忙.如果所有线程池线程都始终保持繁忙,但队列中包含挂起的工作,则线程池将在一段时间后创建另一个辅助线程但线程的数目永远不会超过最大值.超过最大值的线程可以排队,但他们要等到其他线程完成后才启动

什么情况下不要使用线程池

●如果需要使一个任务具有特定优先级

●如果具有可能会长时间运行（并因此阻塞其他任务）的任务
●如果需要将线程放置到单线程单元中（线程池中的线程均处于多线程单元中）
●如果需要永久标识来标识和控制线程，比如想使用专用线程来终止该线程，将其挂起或按名称发现它
System.ThreadingPool类实现了线程池，这是一个静态类，它提供了管理线程的一系列方法
Threading.QueueUserItem方法在线程池中创建一个线程池线程来执行指定方法（用委托WaitCallBack表示），并将该线程排入线程池的队列等待执行。
publc static Boolean QueueUserWorkItem(WaitCallback wc,Object state);

传递参数

调用QueueUserWorkItem时传入的Object类型参数传递到任务过程，可以通过这种方式来向任务过程传递参数。如果任务过程需要多个参数，可以定义包含这些数据的类，并将其强制转换为Object数据类型

线程池实例：

package threadpool;

import ojadb.core.thread.OjadbThreadPool;
import ojadb.core.thread.task.OjadbTask;
import ojadb.exception.OjadbException;

public class BasicTest {
    public static void main(String[] ojadbs) throws OjadbException {
        OjadbThreadPool pool = new OjadbThreadPool(5, 10);
        OjadbTask thread = new OjadbTask("test thread");
        pool.excute(thread);
        System.exit(0);
    }
}

线程池：

package ojadb.core.thread;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import ojadb.core.thread.task.OjadbTask;
import ojadb.core.thread.thread.OjadbThread;
import ojadb.exception.OjadbException;

public class OjadbThreadPool {
    private int minimum;
    private int maximum;
    private LinkedBlockingQueue ojadbThreads;
    private BlockingQueue ojadbTasks;

    public OjadbThreadPool(int min, int max) throws OjadbException {
        if (0 < min && min < max) {
            minimum = min;
            maximum = max;
            ojadbThreads = new LinkedBlockingQueue();
            ojadbTasks = new LinkedBlockingQueue();
            for (int i = 0; i < minimum; i++) {
                OjadbThread worker = new OjadbThread(this, "worker" + i);
                worker.start();
            }
        } else {
            throw new OjadbException("无法建立线程池: 最小值必须大于零,小于最大值");
        }
    }

    public void addWorker(OjadbThread ojadbThreadWorker) {
        ojadbThreads.offer(ojadbThreadWorker);
    }

    public void excute(OjadbTask thread) {
        // 生产任务
        ojadbTasks.offer(thread);

        OjadbThread worker = ojadbThreads.remove();
        if (worker != null) {
            worker.work();
        } else if (ojadbThreads.size() < maximum) {
            OjadbThread newWorker = new OjadbThread(this, "worker" + ojadbThreads.size());
            newWorker.start();
        } else {

        }
        releaseThreads();
    }

    private void releaseThreads() {
        while (ojadbThreads.size() > minimum) {
            OjadbThread worker = ojadbThreads.remove();
            System.out.println(worker + "is killed");
            worker.kill();
        }
    }

    public OjadbTask getTask() {
        if (ojadbTasks.size() == 0)
            return null;
        // 消费任务
        return ojadbTasks.remove();
    }
}

内部线程：

package ojadb.core.thread.thread;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import ojadb.core.thread.OjadbThreadPool;
import ojadb.core.thread.task.OjadbTask;

public class OjadbThread extends Thread {
    private boolean isActived;
    private ReentrantLock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();
    private OjadbThreadPool pool;

    public OjadbThread(OjadbThreadPool pool, String threadName) {
        super(threadName);
        this.pool = pool;
        isActived = true;
    }

    @Override
    public synchronized void start() {
        super.start();
    }

    @Override
    public void run() {
        try {
            lock.lock();

            while (isActived) {
                OjadbTask task = pool.getTask();
                if (task != null) {
                    String result = (String) task.excute();

                    if (result.length() > 0) {
                        pool.addWorker(this);
                        // 懒工人策略 -- 没人招呼工作，绝不睡醒
                        condition.await();
                    }
                } else {
                    pool.addWorker(this);
                    condition.await();
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public void work() {
        try {
            lock.lock();
            condition.signal();
        } finally {
            lock.unlock();
        }
    }

    public void kill() {
        isActived = false;
        work();
    }

    @Override
    public String toString() {
        return getName();
    }
}

用户线程：

package ojadb.core.thread.task;

public class OjadbTask {
    private String threadName;

    public OjadbTask(String threadName) {
        this.threadName = threadName;
    }

    public Object excute() {
        System.out.println("excuting " + threadName);
        return "success";
    }

    public String getName() {
        return threadName;
    }

    public String toString() {
        return threadName;
    }
}