- 目的
了解线程池的知识后，写个线程池实例，熟悉多线程开发，建议看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();
}*/
}
}
}