并发编程之手写一个简单的线程池

前言：

有些人可能对线程池比较陌生，并且更不熟悉线程池的工作原理。所以他们在使用多线程的时候，往往都是通过直接new Thread来实现多线程。但是往往良好多线程的设计大多都是使用线程池去实现，今天主要是跟大家分享如何自己实现一个简单的线程池，帮助理解线程池的工作的原理，以及手动实现的这个线程池存在哪些缺点不足，最后分析JDK源码中的线程池是如何设计来解决这些缺点和不足。

一，为什么要使用线程池

（1）降低资源的消耗。降低线程创建和销毁的资源消耗；

（2）提高响应速度：线程的创建时间为T1，执行时间T2,销毁时间T3，免去T1和T3的时间

（3）提高线程的可管理性。

二，手写一个线程池

1、在手写一个线程池之前，我们应该简单的考虑一下，这个线程的数据结构

（1）线程池中运行线程的个数

（2）线程池中如何保存未处理的任务

（3）线程池中的执行任务和清除任务的方法

2、直接上代码吧

MySelfThreadPool .java

package com.concurrent.pool;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class MySelfThreadPool {
	//默认线程池中的线程的数量
	private static final int WORK_NUM = 5;
	//默认处理任务的数量
	private static final int TASK_NUM = 100;
	
	private int workNum;//线程数量
	
	private int taskNum;//任务数量
	
	private final Set<WorkThread> workThreads;//保存线程的集合
	
	private final BlockingQueue<Runnable> taskQueue;//阻塞有序队列存放任务
	
	
	public MySelfThreadPool() {
		this(WORK_NUM, TASK_NUM);
	}


	public MySelfThreadPool(int workNum, int taskNum) {
		if (workNum <= 0) workNum = WORK_NUM;
		if (taskNum <= 0) taskNum = TASK_NUM;
		taskQueue = new ArrayBlockingQueue<>(taskNum);
		this.workNum = workNum;
		this.taskNum = taskNum;
		workThreads = new HashSet<>();
		//启动一定数量的线程数，从队列中获取任务处理
		for (int i=0;i<workNum;i++) {
			WorkThread workThread = new WorkThread("thead_"+i);
			workThread.start();
			workThreads.add(workThread);
		}
	}
	
	/**
	 * 线程池执行任务的方法，其实就是往BlockingQueue中添加元素
	 * @param task
	 */
	public void execute(Runnable task) {
		try {
			taskQueue.put(task);
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public void destroy() {
		System.out.println("ready close thread pool...");
		if (workThreads == null || workThreads.isEmpty()) return ;
		for (WorkThread workThread : workThreads) {
			workThread.stopWork();
			workThread = null;//help gc
		}
		workThreads.clear();
	}
	
	/**
	 * 线程池中的工作线程，直接从BlockingQueue中获取任务
	 * 然后执行任务而已
	 * blockQueue为阻塞队列
	 *
	 */
	private class WorkThread extends Thread{
		public WorkThread(String name) {
			super();
			setName(name);
		}
		
		@Override
		public void run() {
			while (!interrupted()) {
				try {
					Runnable runnable = taskQueue.take();//获取任务
					if (runnable !=null) {
						System.out.println(getName()+" ready execute:"+runnable.toString());
						runnable.run();//执行任务
					}
					runnable = null;//help gc
				} catch (Exception e) {
					interrupt();
					e.printStackTrace();
				}
			}
		}
		
		
		public void stopWork() {
			interrupt();
		}
	}
	
	
}

到上面，我们就已经实现了一个简单的线程池了，当然功能比较简单，但是对我们理解线程池的工作原理还是很有帮助，下面我们来看看测试程序

TestMySelfThreadPool .java

package com.concurrent.pool;

public class TestMySelfThreadPool {
	
	private static final int TASK_NUM = 50;//任务的个数

	public static void main(String[] args) {
		MySelfThreadPool myPool = new MySelfThreadPool(3,50);
		for (int i=0;i<TASK_NUM;i++) {
			myPool.execute(new MyTask("task_"+i));
		}
		
	}
	
	static class MyTask implements Runnable{
		
		private String name;
		public MyTask(String name) {
			this.name = name;
		}
		
		public String getName() {
			return name;
		}

		public void setName(String name) {
			this.name = name;
		}


		@Override
		public void run() {
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			System.out.println("task :"+name+" end...");
			
		}
		
		@Override
		public String toString() {
			// TODO Auto-generated method stub
			return "name = "+name;
		}
	}
}

3， 程序运行的部分截图：

上面实现的线程池其实非常简单，就是客户端往线程池中的队列中添加任务，然后线程池中的线程一直从队列中去拿任务，拿到就执行，拿不到就一直阻塞。测试程序中就是创建多个任务，然后往线程池中扔。

虽然上面实现的简单的线程池能够实现基本功能，但是还是存在很多的不足

（1）线程池中的线程的个数，不能随着任务进行自动调整，这可能导致线程的浪费，或者导致提交的任务一直无法执行，导致应用崩溃

（2）当无法处理任务的时候，没有很好的方式一直去处理，而是让它一直阻塞

针对上面的问题，我们来看看jdk源码中的线程池是如何去处理的

三，ThreadPoolExecutor原理分析

先看下ThraedPoolExecutor的构造方法（参数最多的那个）

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {}

1，参数解析：

int corePoolSize：线程池中核心线程数，< corePoolSize，就会创建新线程，= corePoolSize，这个任务就会保存到BlockingQueue，如果调用prestartAllCoreThreads（）方法就会一次性的启动corePoolSize个数的线程。

int maximumPoolSize：允许的最大线程数，BlockingQueue也满了，< maximumPoolSize时候就会再次创建新的线程

long keepAliveTime：线程空闲下来后，存活的时间，这个参数只在> corePoolSize才有用

TimeUnit unit：存活时间的单位值（秒、毫秒…）

BlockingQueue<Runnable> workQueue：保存任务的阻塞队列

ThreadFactory threadFactory：创建线程的工厂，给新建的线程赋予名字

RejectedExecutionHandler handler：饱和策略

AbortPolicy：直接抛出异常，默认；

CallerRunsPolicy：用调用者所在的线程来执行任务

DiscardOldestPolicy：丢弃阻塞队列里最老的任务，队列里最靠前的任务

DiscardPolicy：当前任务直接丢弃

实现自己的饱和策略，实现RejectedExecutionHandler接口即可

2，图解

看了上面的参数解析，其实基本上已经了解了它的处理的逻辑了，下面用一个图来解释一下

（1）当线程池中的线程个数<corePool，每次进来新的任务，就启动一个线程去处理这个任务

（2）当线程池中的线程个数>=corePool，这时每次进来的新的任务，就会加入到队列中。

（3）当线程池中的线程个数>=corePool但是<maxiMumPool，并且队列也满了（有界队列），这个时候再来新的任务，就会继续创建新的线程去处理。这个时候创建的线程，当线程空闲下来的时候到keepAliveTime的时间就会销毁（线程资源宝贵啊）

（4）当线程池中的线程数>=maximumPool时，这个时候再来新的任务，就可以选择拒绝的机制

3、代码解析

接下来看看代码验证一下上面的描述把，看下面这一段就行了

好了，就分享到这里把，欢迎指正！