Java多线程---线程池源码分析及其实现

之前Java线程池的文章都是关于基本知识和JUC下的类。这篇主要来说一下线程池，并自己来实现一个线程池。

一.线程池介绍

学习过程中会遇到各种池，有线程池，数据库连接池，内存池，常量池等等。下面来一次介绍。

线程池： 用来管理线程的一个集合(池)，作用是用来提高线程的使用效率。如果一个线程的创建和销毁的成本比运行该线程里面的程序的成本要高，则就需要用到线程池。在线程池中，里面的线程可以重用，而不用每一个任务都创建一个线程。Java5引入了Executors类产生线程池。

数据库连接池： 程序要和数据库通信就需要建立一个连接，这个连接的建立和关闭时及其耗费系统资源，数据库连接池就是用来管理这些连接的。它的原理和线程池一样，只不过线程换成了连接。我之前使用过的C3P0连接池，它是一个开源的JDBC连接池，Hibernate和Spring使用它。

内存池： 内存在C/C++上是手动管理的，对象的新建和销毁是要手动在内存堆上分配和释放，这个会消耗系统资源，如果应用程序频繁地在堆上分配和释放内存，则会导致性能的损失。并且会使系统中出现大量的内存碎片，降低内存的利用率。这是就要用到内存池来对内存的分配和释放进行管理。由于Java是自动管理内存的，所以没有办法实现内存池。

常量池： 常量池是JVM的一块特殊的内存空间，属于Java内存中的方法去，用于存放常量(8种基本)和符号引用。它同上面三种不是一类的。

二.Java线程池类
Java5开始，java内建支持线程池。JUC包下的Executors工厂类（工具类）用来产生线程池。使用它的子接口来定义线程池，通过静态方法来创建不同的线程池。有固定数目大小的线程池，有可延迟执行的线程池，有具有缓存功能的线程池等等。两个子接口ExecutorService,ScheduledExecutorService继承了接口Executor。
下面我们来看一下java内部是如何实现线程池源码的。这里就拿个简单的固定线程池来说明，其他类型的也差不多。下面是一个创建并运行线程池的例子：

public static void main(String [] args){
ExecutorService pool=Executors.newFixedThreadPool(5);//创建具有5个固定线程的线程池
pool.submit(new MyThread());  //提交任务，MyThread是一个实现Runnable接口或者Callable接口的类
pool.submit(new MyThread());
}

首先是创建一个ExecutorService线程池，我们看看源码是：

public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}

它其实是调用了ThreadPoolExecutor类，我们也可以直接创建这个类。它的构造方法如下：

public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}

创建固定的线程数参数nThreads，在上面是将corePoolSizes和maximumPoolSize都设为相同的值。workQueue是一个任务队列，它是用来放需要执行的任务，任务是要实现Runnable接口的，它的实现是LinkedBlockingQueue<Runnable>()，它叫链表阻塞队列，意思是在获取任务时，如果里面没有任务，就会造成阻塞。threadFactory是用来创建Thread线程的。其他的参数在这里没啥作用。再来看看ExecutorService.submit()方法，它用来提交任务。它其实是调用了它实现类AbstractExecutorService的submit方法：

public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Object> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}

它创建了一个FutureTask类，将提交的任务task放在了FutureTask类中，这个类上一篇文章说过，就是在线程中执行任务。然后将执行的任务调用execute()方法。下面看execute方法。这个方法，在ThreadPoolExecutor实现。它的定义是这样的public class ThreadPoolExecutor extends AbstractExecutorService。它才是整个线程池的关键类。下面看execute()方法：

public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
if (runState == RUNNING && workQueue.offer(command)) {
if (runState != RUNNING || poolSize == 0)
ensureQueuedTaskHandled(command);
}
else if (!addIfUnderMaximumPoolSize(command))
reject(command); // is shutdown or saturated
}
}

第一个if没啥好解释，如果任务是null直接抛出异常。
第二个if里面有两个判断，第一个判断：poolSize>=corePoolSize 它们是int类型，如果当前的线程数大于定义的core线程数；或者 第二个判断：（主要如果第一个判断是true则不用进行第二个判断），就是当当前线程数小于定义的线程数时，才调用addIfUnderCorePoolSize()方法，一开始，当前线程数是小于定义的线程数，所以来看看addIfUnderCorePoolSize方法：

private boolean addIfUnderCorePoolSize(Runnable firstTask) {
Thread t = null;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (poolSize < corePoolSize && runState == RUNNING)
t = addThread(firstTask);
} finally {
mainLock.unlock();
}
return t != null;
}

里面还用到了Lock类进行同步处理，它里面又进行了一次线程输入判断，当运行状态为RUNNING时，才调用addThread(firstTask)方法，它是创建任务线程的方法，也就是在addThread方法中创建Thread类，我们来看方法：

private Thread addThread(Runnable firstTask) {
Worker w = new Worker(firstTask);
Thread t = threadFactory.newThread(w);
boolean workerStarted = false;
if (t != null) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
w.thread = t;
workers.add(w);
int nt = ++poolSize;
if (nt > largestPoolSize)
largestPoolSize = nt;
try {
t.start();
workerStarted = true;
}
finally {
if (!workerStarted)
workers.remove(w);
}
}
return t;
}

里面有一个类Worker：任务的执行者，它是ThreadPoolExecutor类的一个内部类，用于执行任务，它将Thread作为它的一个属性。然后将定义的w对象加入到workers中。workers是一个HashSet<Worker>类，它用来存放Worker。然后将新建的Thread开启，用了start()。Thread任务执行的是w的run()方法。Worker的run方法如下：

public void run() {
try {
hasRun = true;
Runnable task = firstTask;
firstTask = null;
while (task != null || (task = getTask()) != null) {
runTask(task);
task = null;
}
} finally {
workerDone(this);
}
}

它里面是一个while循环，先从firstTask运行，如果为null，则调用getTask()方法，获取任务。然后在用runTask方法执行任务。getTask方法如下：

Runnable getTask() {
for (;;) {
try {
int state = runState;
if (state > SHUTDOWN)
return null;
Runnable r;
if (state == SHUTDOWN)  // Help drain queue
r = workQueue.poll();
else if (poolSize > corePoolSize || allowCoreThreadTimeOut)
r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
else
r = workQueue.take();
if (r != null)
return r;
if (workerCanExit()) {
if (runState >= SHUTDOWN) // Wake up others
interruptIdleWorkers();
return null;
}
// Else retry
} catch (InterruptedException ie) {
// On interruption, re-check runState
}
}

它也是一个循环，关键是这句 r = workQueue.take()，就是从阻塞队列中获取任务并在队列中删除获取的任务(它保证了线程的安全)，如果有则获取成功，结束for循环返回，如果队列里面没有任务则阻塞，直到里面有新任务才返回。然后就执行runTask(task)这就，它才是真正执行任务的程序。runTask里面的关键代码如下：

try {
task.run();
ran = true;
afterExecute(task, null);
++completedTasks;
} catch (RuntimeException ex) {
if (!ran)
afterExecute(task, ex);
throw ex;
}

是调用了任务的run方法。这样一个任务就完成了。 Worker的workerDone方法表明，如果这个任务执行者结束了while循环，代表了这个线程会被删除，就由workerDone方法来完成。这就是一个基本的线程池实现。

再来说一下，线程池就是开启了多个线程，把它们放入到HashSet中，需要执行的任务放在Queue队列中，多个线程就不断的获取队列中的任务进行执行。用Lock类来确保数据的安全性，如线程池的数量，各个int类型的参数，在获取执行任务时并没有加锁，而是用到了安全的阻塞队列，它是LinkedBlockingQueue<Runnable>()，它确保调用的take()方法是安全的，因此多个线程在获取任务的时候不会发生重读和死锁。

三.自己编写Java线程池

通过上面对java自带线程池源码的分析，知道了线程池的内部构造，这样自己写一个线程池也就不难了。我自己就实现一个简单的有固定数目的线程池。线程池采用数组里存放线程，采用LinkedList存放任务，获取任务的时候使用synchronized进行同步处理确保线程安全。任务执行线程MyThreadPool继承了Thread，这样就不用把Thread类作为属性了，代码如下：

class MyThreadPool{
private int threadsize;  //线程的个数
private MyThread[] threadpool;	       //存放线程的线程池
private LinkedList<Runnable> queue;    //任务队列
private boolean state=true;  //true代表线程池是开启的，如果是false则是关闭的
public MyThreadPool(int size){
threadsize=size;
queue=new LinkedList<Runnable>();
threadpool=new MyThread[threadsize];
for(int i=0;i<size;i++){
threadpool[i]=new MyThread();
threadpool[i].start();  //创建完之后随即开启线程
}
}
public void submit(Runnable task){  //提交任务，就是向队列中添加任务
synchronized(queue){
queue.add(task);
queue.notify();
}
}
public void shutdown(){  //关闭线程池，需要同步，并调用notefyAll方法唤醒阻塞的线程
synchronized(queue){
state=false;
queue.notifyAll();
}
}
private class MyThread extends Thread{
private Runnable task;
@Override
public void run(){
while(state){   //当state是true时才运行，否则结束
synchronized(queue){
while(queue.isEmpty() && state){
try{
queue.wait();//阻塞该线程，直到有新任务加入队列或者关闭线程池
}
catch(Exception e){
e.printStackTrace();
}
}
if(!queue.isEmpty()) //再判断一次queue是否有数据，否则可能出现NoSuchElementException错误
task=queue.pop();
}  //结束同步互斥
if(task!=null && state){
task.run();  //运行任务
}
}
}
}
}		 

下面对自己创建的线程池进行测试，代码如下： 执行10000次的自增操作。第一个就创建了10000个线程，每个线程加1，第二个创建了有20个线程的线程池来完成。对比结果，线程池的作用还是蛮明显的。

public class di implements Runnable
{
private static int count=0;
public synchronized void run(){
count++;
}
public static void main(String[] args) throws IOException{
//下面来对我自建的线程池进行测试。用对比实验
//首先是没用线程池的情况，创建10000个线程,每个线程完成自增1的任务,直到对象的count=10000
di test1=new di();
long start=System.currentTimeMillis();
for(int i=0;i<10000;i++){
new Thread(test1).start();
}
long end=0;
while(true){
if(test1.count>=10000){
end=System.currentTimeMillis();
System.out.println("没使用线程池，花费时间："+(end-start));
break;
}
}
//采用线程池
di test2=new di();
MyThreadPool my=new MyThreadPool(20);
start=System.currentTimeMillis();
for(int i=0;i<10000;i++){
my.submit(test2);
}
while(true){
if(test2.count>=10000){
end=System.currentTimeMillis();
System.out.println("使用线程池，花费时间："+(end-start));
break;
}
}
my.shutdown();
}
}

结果：