线程池简单实现

Posted 2020-10-06 insaneXs

实现了一个简化版的线程池。

实现线程池的关键有两个：一是阻塞队列，用于任务的存取，二是内部的线程对象如何持续性的执行任务，并在空闲时被回收。

线程池代码：

  1 package learnConcurrent;
  2 
  3 import java.util.ArrayList;
  4 import java.util.Collection;
  5 import java.util.LinkedList;
  6 import java.util.List;
  7 import java.util.concurrent.ArrayBlockingQueue;
  8 import java.util.concurrent.BlockingQueue;
  9 import java.util.concurrent.Callable;
 10 import java.util.concurrent.ExecutionException;
 11 import java.util.concurrent.ExecutorService;
 12 import java.util.concurrent.Future;
 13 import java.util.concurrent.TimeUnit;
 14 import java.util.concurrent.TimeoutException;
 15 import java.util.concurrent.atomic.AtomicBoolean;
 16 import java.util.concurrent.atomic.AtomicInteger;
 17 import java.util.concurrent.locks.ReentrantLock;
 18 
 19 public class MyThreadPool implements ExecutorService{
 20     //线程队列
 21     private List<Worker> workers;
 22     //任务队列
 23     private BlockingQueue<Runnable> rQueue;
 24     //线程池核心大小
 25     private int corePoolSize;
 26     //线程池最大大小
 27     private int maxPoolSize;
 28     //空闲线程最长存活时间
 29     private int keepAliveTime = 5;
 30 
 31     private static final int ALIVE = 0;
 32     
 33     private static final int SHUTDOMN = 1;
 34     
 35     private AtomicInteger state = new AtomicInteger(ALIVE);
 36     
 37     private ReentrantLock lock = new ReentrantLock();
 38     
 39     public MyThreadPool(int corePoolSize, int maxPoolSize){
 40         this.corePoolSize = corePoolSize;
 41         this.maxPoolSize = maxPoolSize;
 42         
 43         this.workers = new LinkedList<Worker>();
 44         //阻塞队列,最大容量为maxPoolSize
 45         this.rQueue = new ArrayBlockingQueue<Runnable>(maxPoolSize, true);
 46     }
 47     
 48     @Override
 49     public void execute(Runnable command) {
 50         //FIXME size在获取时和判断时 可能发生改变
 51         lock.lock();
 52         int size = workers.size();
 53         if(size < corePoolSize){//当线程池线程数小于核心数量时，增加线程
 54             addWorker();
 55         }else if(size < maxPoolSize && !rQueue.isEmpty()){//当线程大于核心数量且任务队列中任务排队时，增加线程
 56             addWorker();
 57         }
 58         lock.unlock();
 59         
 60         if(!isShutdown()){
 61             rQueue.offer(command);
 62         }
 63     }
 64     
 65     @Override
 66     public void shutdown() {
 67         //关闭线程池的简单实现,设置状态让任务队列不在接受任务,线程也会因为超时被回收
 68         //缺点时空闲的线程资源得不到立即释放
 69         state.set(SHUTDOMN);
 70     }
 71     
 72     /**
 73      * 立即停止线程池，试图停止正在活动的线程,返回还在等待的任务列表
 74      */
 75     @Override
 76     public List<Runnable> shutdownNow() {
 77         if(isShutdown())
 78             return null;
 79         state.set(SHUTDOMN);
 80         lock.lock();
 81         List<Runnable> restRunnable = new ArrayList<Runnable>();
 82         while(!rQueue.isEmpty()){
 83             restRunnable.add(rQueue.poll());
 84         }
 85         for(Worker w : workers){
 86             w.interrupt();
 87         }
 88         lock.unlock();
 89         return restRunnable;
 90     }
 91 
 92     @Override
 93     public boolean isShutdown() {
 94         return state.get() == ALIVE;
 95     }
 96 
 97     @Override
 98     public boolean isTerminated() {
 99         return isShutdown() && rQueue.isEmpty();
100     }
101 
102     @Override
103     public boolean awaitTermination(long timeout, TimeUnit unit)
104             throws InterruptedException {
105         // TODO Auto-generated method stub
106         return false;
107     }
108 
109     @Override
110     public <T> Future<T> submit(Callable<T> task) {
111         // TODO Auto-generated method stub
112         return null;
113     }
114 
115     @Override
116     public <T> Future<T> submit(Runnable task, T result) {
117         // TODO Auto-generated method stub
118         return null;
119     }
120 
121     @Override
122     public Future<?> submit(Runnable task) {
123         return null;
124     }
125 
126     @Override
127     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
128             throws InterruptedException {
129         return null;
130     }
131 
132     @Override
133     public <T> List<Future<T>> invokeAll(
134             Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
135             throws InterruptedException {
136         return null;
137     }
138 
139     @Override
140     public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
141             throws InterruptedException, ExecutionException {
142         return null;
143     }
144 
145     @Override
146     public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
147             long timeout, TimeUnit unit) throws InterruptedException,
148             ExecutionException, TimeoutException {
149         return null;
150     }
151     
152     private Runnable getTask(){
153         Runnable r = null;
154         try {
155             r = rQueue.poll(keepAliveTime, TimeUnit.SECONDS);
156         } catch (InterruptedException e) {
157             // TODO Auto-generated catch block
158             e.printStackTrace();
159         }
160         return r;
161     }
162     
163     private void addWorker(){
164         Worker w = new Worker();
165         w.start();
166         lock.lock();
167         workers.add(w);
168         lock.unlock();
169     }
170     
171     private void removeWorker(Worker w){
172         lock.lock();
173         workers.remove(w);
174         lock.unlock();
175     }
176     
177     class Worker extends Thread{
178         
179         private AtomicBoolean isAlive = new AtomicBoolean(true);
180         
181         private Runnable task;
182         
183         
184         @Override
185         public void run() {
186             while(isAlive.get()){
187                 //阻塞一定时间,超时则回收该线程
188                 task = getTask();
189                 if(task != null){
190                     task.run();
191                 }else{
192                     isAlive.set(false);
193                     
194                 }
195                 task = null;
196             }
197             System.out.println("remove worker");
198             removeWorker(this);
199         }
200         
201     }
202     
203     
204 }

测试代码：

 1 package learnConcurrent;
 2 
 3 
 4 public class ThreadPoolTest {
 5     static int taskNo = 0;
 6     public static void main(String[] args) throws InterruptedException {
 7         MyThreadPool pool = new MyThreadPool(2, 5);
 8         
 9         for(int i=0; i< 50; i++){
10             Task task = new Task(taskNo++);
11             pool.execute(task);
12             Thread.sleep((int)(Math.random() * 1000));
13         }
14         
15     }
16     
17 }
18 
19 class Task implements Runnable{
20     String str;
21     public Task(int taskNo){
22         str = "TaskNo:" + taskNo;
23     }
24     @Override
25     public void run() {
26         System.out.println(str + " start work ");
27         //DO SOMETHING
28         try {
29             Thread.sleep((int)(Math.random() * 1000));
30         } catch (InterruptedException e) {
31             // TODO Auto-generated catch block
32             e.printStackTrace();
33         }
34         
35         System.out.println(str + " done ");
36     }
37     
38 }

虽然是继承了ExecutorService对象，但是只实现了几个接口，设计上也可能有未考虑到的问题。

测试代码也很简陋，仅供参考。