线程池的实现原理

  1 package com.test.ThreadPool;
  2 
  3 import java.util.LinkedList;
  4 import java.util.List;
  5 import java.util.concurrent.atomic.AtomicInteger;
  6 /**
  7  * 线程池的实现原理
  8  * @author Flynn
  9  *
 10  */
 11 public class ThreadPool {
 12     private static ThreadPool pool = null;
 13     
 14     private final int initNum ;    //初始化线程数量
 15     private volatile AtomicInteger quenLen = new AtomicInteger(0);    //任务队列长度，volatile修饰的原子操作不必加入同步模块
 16     private List<Runnable> taskQuen = new LinkedList<Runnable>() ;    //待处理任务队列，涉及其的操作要做线程安全处理（使用linkedlist是因为任务队列插入删除操作比较频繁）
 17     private workThread[] threads;
 18     private ThreadPool(int num){
 19         this.initNum = num ;
 20         threads = new workThread[initNum] ;
 21         for(int i=0;i<initNum;i++){
 22             threads[i] = new workThread() ;
 23             threads[i].start() ;
 24         }
 25     }
 26     /**
 27      * 获取单例
 28      * @param num
 29      * @return
 30      */
 31     public static ThreadPool getPool(int num){
 32         if(pool == null){
 33             synchronized (ThreadPool.class) {
 34                 if(pool == null)
 35                     pool = new ThreadPool(num) ;
 36             }
 37         }
 38         return pool ;
 39     }
 40     /**
 41      * 添加执行任务
 42      * @param runs
 43      */
 44     public void execute(Runnable...runs){
 45         for(Runnable r : runs){
 46             //操作共享资源任务队列 需要同步
 47             synchronized (taskQuen) {
 48                 taskQuen.add(r) ;
 49                 taskQuen.notifyAll() ;
 50             }
 51             quenLen.incrementAndGet() ;
 52         }
 53     }
 54     public int getTaskNum(){
 55         return quenLen.intValue() ;
 56     }
 57     public void detory(){
 58         //如果任存在任务，释放锁对象并等待20ms再次进入可执行状态
 59         while(!taskQuen.isEmpty()){
 60             try {
 61                 Thread.sleep(20) ;
 62             } catch (InterruptedException e) {
 63                 // TODO Auto-generated catch block
 64                 e.printStackTrace();
 65             }
 66         }
 67         for(int i=0;i<initNum;i++){
 68             threads[i].stopWork() ;
 69             threads[i] = null ;    //设置为null便于垃圾回收
 70         }
 71         pool = null ;
 72         taskQuen.clear() ;
 73     }
 74     /**
 75      * 内部线程类
 76      * @author Flynn
 77      *
 78      */
 79     private class workThread extends Thread{
 80         private boolean isRunning = true ;    //通过此标志放行线程，也就是说让线程执行完并销毁
 81         @Override
 82         public void run() {
 83             // TODO Auto-generated method stub
 84             Runnable run = null ;
 85             while(isRunning){
 86                 synchronized (taskQuen) {
 87                     //如果没有任务，释放锁对象并等待20ms再次进入可执行状态
 88                     while(isRunning && taskQuen.isEmpty()){
 89                         try {
 90                             taskQuen.wait(20) ;
 91                         } catch (InterruptedException e) {
 92                             // TODO Auto-generated catch block
 93                             e.printStackTrace();
 94                         }
 95                     }
 96                     if(!taskQuen.isEmpty()){
 97                         //将任务拿出执行，但不是执行成功如否
 98                         run = taskQuen.remove(0);
 99                         quenLen.getAndDecrement();
100                     }
101                 }
102                 //任务执行，因为没有涉及到共享资源，不必放入同步模块
103                 if(run!=null){
104                     run.run() ;
105                 }
106                 run = null ;
107             }
108         }
109         public void stopWork(){
110             this.isRunning = false ;
111         }
112     }
113 }

 1 package com.test.ThreadPool;
 2 
 3 import java.util.concurrent.atomic.AtomicInteger;
 4 
 5 public class TestMain {
 6 
 7     /**
 8      * @param args
 9      * @throws InterruptedException 
10      */
11     public static void main(String[] args) throws InterruptedException {
12         // TODO Auto-generated method stub
13         ThreadPool pool = ThreadPool.getPool(4) ;
14         new TestMain.myRun() ;
15         pool.execute(new myRun(),new myRun(),new myRun(),new myRun()) ;
16         pool.execute(new myRun(),new myRun(),new myRun(),new myRun()) ;
17         pool.execute(new myRun(),new myRun(),new myRun(),new myRun()) ;
18         pool.execute(new myRun(),new myRun(),new myRun(),new myRun()) ;
19         pool.detory() ;
20     }
21     static class myRun implements Runnable{
22         private static volatile AtomicInteger len = new AtomicInteger(0);    //任务队列长度，原子操作
23         @Override
24         public void run() {
25             // TODO Auto-generated method stub
26             System.err.println("任务--"+len.incrementAndGet()+"--执行了");
27         }
28     }
29 }

 1 任务--4--执行了
 2 任务--3--执行了
 3 任务--1--执行了
 4 任务--2--执行了
 5 任务--7--执行了
 6 任务--6--执行了
 7 任务--5--执行了
 8 任务--10--执行了
 9 任务--9--执行了
10 任务--8--执行了
11 任务--13--执行了
12 任务--12--执行了
13 任务--11--执行了
14 任务--16--执行了
15 任务--15--执行了
16 任务--14--执行了