java线程池实例

• 目的

        了解线程池的知识后，写个线程池实例，熟悉多线程开发，建议看jdk线程池源码，跟大师比，才知道差距啊O(∩_∩)O

---

•   线程池类

package thread.pool2;

import java.util.LinkedList;

public class ThreadPool {
    //最大线程数
    private int maxCapacity;
    //初始线程数
    private int initCapacity;
    //当前线程数
    private int currentCapacity;
    //线程池需要执行的任务
    private LinkedList<Task> tasks;
    //当前处于等待的线程数
    private int waitThreadNum = 0;
    //线程池中线程数超过初始数量时，此时有线程执行完任务,但是没有后续的任务执行,则会等待一段时间后，该线程才销毁
    //destroyTime小于或等于0时,线程立即消费,大于0,则等待设置的时间
    private int destroyTime = 0;
    
    public ThreadPool(int initCapacity,int maxCapacity, int destroyTime) {
        if(initCapacity > maxCapacity) {
            //初始线程数不能超过最大线程数,当然此处可以抛出异常,提示不允许这么设置
            initCapacity = maxCapacity;
        }
        this.maxCapacity = maxCapacity;
        this.initCapacity = initCapacity;
        this.currentCapacity = initCapacity;
        this.tasks = new LinkedList<Task>();
        this.waitThreadNum = initCapacity;
        this.destroyTime = destroyTime;
    }
    /**
     * 向线程池中添加任务,如果线程数不够,则增加线程数,但线程数总量不能超过给定的最大线程数
     * @param task
     */
    public synchronized void addTask(Task task) {
        tasks.add(task);
        addThread();
        notifyAll();
    }
    /**
     * 从线程池中取出任务,如果没有任务,则当前线程处于等待状态
     * @return
     * @throws InterruptedException
     */
    public synchronized Task getTask() throws InterruptedException {
        while(tasks.isEmpty()) {
            wait();
        }
        //取出第一个任务的同时将第一个任务移除
        return tasks.pollFirst();
    }
    /**
     * 判断线程池中任务列表是否为空
     * @return
     */
    public synchronized boolean isEmpty() {
        return tasks.isEmpty();
    }
    /**
     * 活跃线程数加1
     */
    public synchronized void addWaitThreadNum(int num) {
        waitThreadNum += num;
    }
    /**
     * 活跃线程数减1
     */
    public synchronized void reduceWaitThreadNum(int num) {
        waitThreadNum -= num;
    }
    
    /**
     * 启动线程池
     */
    public void execute() {
        System.out.println(initCapacity);
        for(int i = 0; i < initCapacity; i++) {
            (new Thread(new InnerThread(this, "thread"+ i))).start();
        }
    }
    /**
     * 如果当前线程数大于初始线程数,则关闭当前线程,否则当前线程处于等待状态
     * @return
     * @throws InterruptedException 
     */
    public synchronized boolean waitOrClose(int tmp) throws InterruptedException {
        System.out.println(currentCapacity + ":" + initCapacity);
        //线程退出前,等待一段时间,防止线程频繁创建和销毁线程
        if(destroyTime > 0) {
            wait(destroyTime);
        }
        if(currentCapacity > initCapacity && tasks.isEmpty()) {
            currentCapacity--;
            System.out.println("任务执行完后,当前线程数：" + currentCapacity);
            return false;
        }
        System.out.println("线程等待结束");
        addWaitThreadNum(tmp);
        wait();
        return true;
    }
    
    /**
     * 当线程池内线程数不够时,如果有任务在等待处理,同时当前线程都处于非等待状态,
     * 则增加线程池中线程数,但不能超过线程池中最大线程数
     */
    public synchronized void addThread() {
        System.out.println("当前线程数：" + currentCapacity + "最大线程数：" + maxCapacity + "等待线程数" + waitThreadNum);
        if(currentCapacity < maxCapacity && waitThreadNum == 0) {
            //每添加一个线程,当前线程数加1
            currentCapacity++;
            //每添加一个线程,相当于线程池中多了一个等待的线程
            waitThreadNum++;
            System.out.println("当前线程数为：" + currentCapacity);
            new Thread(new InnerThread(this, "thread" + (currentCapacity-1))).start();
        }
    }
    /**
     * 线程池中单个线程对象
     * @author yj
     *
     */
    private class InnerThread implements Runnable {
        
        private ThreadPool threadPool;
        private String threadName;
        
        public InnerThread(ThreadPool threadPool, String threadName) {
            this.threadPool = threadPool;
            this.threadName = threadName;
        }

        @Override
        public void run() {
            try {
                while(true){
                    int addWait = 0;
                    int resuceWait = 1;
                    //不等于空,则处理任务
                    while(!threadPool.isEmpty()) {
                        threadName = Thread.currentThread().getName();
                        reduceWaitThreadNum(resuceWait);
                        Task task = threadPool.getTask();
                        task.execute(threadName);
                        try {
                            Thread.sleep(9000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println(threadName + "对"+task.getTaskName()+"+任务进行了处理");
                        //只有处理任务后回到等待状态的线程才将waitThreadNum加1
                        addWait = 1;
                        //如果不跳出循环,则等待线程数不减少
                        resuceWait = 0;
                    }
                    //等于空,则等待任务或关闭当前线程
                    if(threadPool.waitOrClose(addWait)) {    
                        System.out.println(threadName + "处于等待状态");
                        continue;
                    }
                    //关闭线程
                    break;
                }                
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

---

• 任务类

package thread.pool2;

public class Task{
    
    private String taskName;
    
    public String getTaskName() {
        return taskName;
    }

    public Task(String taskName) {
        this.taskName = taskName;
    }

    public void execute(String threadName) {
        System.out.println(threadName + "开始执行任务为" + taskName);
        /*try {
            Thread.sleep(9000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }*/
        System.out.println(threadName + "执行" + taskName + "任务完成");
    }

}

---

• 测试类

package thread.pool2;

public class ThreadPoolTest {

    public static void main(String[] args) {
        ThreadPool threadPool = new ThreadPool(3, 10, 1100);
        threadPool.execute();
        for(int i = 0; i < 50; i++) {
            int random = (int) (Math.random() * 1000);
            threadPool.addTask(new Task("task"+random));
            /*try {
                //每个1秒向线程池中添加任务
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }*/
        }    
    }

}