pooledthread.java

Java线程池完整代码

package com.usstinfo.skyeyecore.threadpool;

import java.util.Collection;
import java.util.Iterator;
import java.util.Vector;

/** *//**
 * 接受线程池管理的线程
 * @author Leo Chang QQ:278475286
 * 2008-8-8
 */
public class PooledThread extends Thread {
   
    protected Vector tasks = new Vector();
    protected boolean running = false;
    protected boolean stopped = false;
    protected boolean paused = false;
    protected boolean killed = false;
    private ThreadPool pool;
   
    public PooledThread(ThreadPool pool){
        this.pool = pool;
        pool.threadCreated(this);
    }
   
    public void putTask(ThreadTask task) {
        tasks.add(task);
    }
   
    public void putTasks(ThreadTask[] tasks) {
        for(int i=0; i<tasks.length; i++)
            this.tasks.add(tasks[i]);
    }

    protected ThreadTask popTask() {
        if(tasks.size() > 0)
            return (ThreadTask)tasks.remove(0);
        else
            return null;
    }
   
    public boolean isRunning() {
        return running;
    }
    
    public boolean isTaskQueueEmpty() {
    	return tasks.isEmpty();
    }
   
    public void stopTasks() {
        stopped = true;
    }
   
    public void stopTasksSync() {
        stopTasks();
        while(isRunning()) {
            try {
                sleep(5);
            } catch (InterruptedException e) {
            }
        }
    }
   
    public void pauseTasks() {
        paused = true;
    }
   
    public void pauseTasksSync() {
        pauseTasks();
        while(isRunning()) {
            try  {
                sleep(5);
            } catch (InterruptedException e)  {
            }
        }
    }
   
    public void kill() {
        if(!running)
            interrupt();
        else
            killed = true;
    }
   
    public void killSync() {
        kill();
        while(isAlive()) {
            try  {
                sleep(5);
            } catch (InterruptedException e)  {
            }
        }
    }
   
    public synchronized void activeThread() {
    	// 把线程从等待状态中激活，开始任务循环。
        running = true;
        pool.threadIdleStateChange(this, false);
        this.notify();
    }
   
    public synchronized void run() {
        try {
            while(true) {
            	// 处理控制线程命令先。
            	if(killed) {
            		// 杀掉线程：向线程池任务队列转移自己所有任务，退出。
                	killed = false;
                	returnTaskToPool();
                	break;
            	}
            	if(stopped) {
            		// 停止线程：向线程池任务队列转移自己所有任务，然后等待。
                    stopped = false;
                    returnTaskToPool();
                    running = false;
                }
                if(paused) {
                	// 暂停线程，保持自己的任务队列，然后等待。
                    paused = false;
                    running = false;
                }

            	ThreadTask task;
            	// 优先执行Thread自己的任务队列,再执行线程池中的任务队列。
            	if(running&&(task = pool.getTaskToRun())!=null){
            		task.run();
            		pool.taskCompleted();
            	}else if(running&&(task = popTask())!=null){
       			 	task.run();
       			 	// 该线程的任务包执行完毕，表示线程池中的一个大任务包执行完成。
            		if(tasks.size()==0)
           			 	pool.taskCompleted();
            	}else{
            		running = false;
            		pool.threadIdleStateChange(this, true);
                    this.wait();
            	}
            }
        }catch(InterruptedException e) {
        	// 在kill()方法中调用interrupt()方法，导致抛出此异常，结束线程。
        	// 非调试状态，注释掉让控制台输出clean
        	//e.printStackTrace();
            return;
        }finally{
        	returnTaskToPool();
        	tasks = null;
        	pool.threadDead(this);
        }
    }
    
    private void returnTaskToPool() {
    	 pool.taskReturnedFromThread(tasks);
    	 tasks.clear();
    }
}