线程池简单实现

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

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

线程池代码：

package learnConcurrent;

import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

public class MyThreadPool implements ExecutorService{
    //线程队列
    private List<Worker> workers;
    //任务队列
    private BlockingQueue<Runnable> rQueue;
    //线程池核心大小
    private int corePoolSize;
    //线程池最大大小
    private int maxPoolSize;
    //空闲线程最长存活时间
    private int keepAliveTime = 60;

    private static final int ALIVE = 0;
    
    private static final int SHUTDOMN = 1;
    
    private int state = ALIVE;
    
    private ReentrantLock lock = new ReentrantLock();
    
    public MyThreadPool(int corePoolSize, int maxPoolSize){
        this.corePoolSize = corePoolSize;
        this.maxPoolSize = maxPoolSize;
        
        this.workers = new LinkedList<Worker>();
        //阻塞队列,最大容量为maxPoolSize
        this.rQueue = new ArrayBlockingQueue<Runnable>(maxPoolSize, true);
    }
    
    @Override
    public void execute(Runnable command) {
        if(isShutdown())
            return;
        //FIXME size在获取时和判断时 可能发生改变
        lock.lock();
        int size = workers.size();
        if(size < corePoolSize){//当线程池线程数小于核心数量时，增加线程
            addWorker();
        }else if(size < maxPoolSize && !rQueue.isEmpty()){//当线程大于核心数量且任务队列中任务排队时，增加线程
            addWorker();
        }
        lock.unlock();
        
        rQueue.offer(command);
    }
    
    @Override
    public void shutdown() {
        //关闭线程池的简单实现,设置状态让任务队列不在接受任务,线程也会因为超时被回收
        //缺点时空闲的线程资源得不到立即释放
        lock.lock();
        state = SHUTDOMN;
        lock.unlock();
    }
    
    /**
     * 立即停止线程池，试图停止正在活动的线程,返回还在等待的任务列表
     */
    @Override
    public List<Runnable> shutdownNow() {
        if(isShutdown())
            return null;
        lock.lock();
        state = SHUTDOMN;
        List<Runnable> restRunnable = new ArrayList<Runnable>();
        while(!rQueue.isEmpty()){
            restRunnable.add(rQueue.poll());
        }
        for(Worker w : workers){
            w.interrupt();
        }
        lock.unlock();
        return restRunnable;
    }

    @Override
    public boolean isShutdown() {
        lock.lock();
        boolean res = state != ALIVE;
        lock.unlock();
        return res;
    }

    @Override
    public boolean isTerminated() {
        return isShutdown() && rQueue.isEmpty();
    }

    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException {
        // TODO Auto-generated method stub
        return false;
    }

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public <T> Future<T> submit(Runnable task, T result) {
        // TODO Auto-generated method stub
        return null;
    }

    @Override
    public Future<?> submit(Runnable task) {
        return null;
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
            throws InterruptedException {
        return null;
    }

    @Override
    public <T> List<Future<T>> invokeAll(
            Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
            throws InterruptedException {
        return null;
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
            throws InterruptedException, ExecutionException {
        return null;
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
            long timeout, TimeUnit unit) throws InterruptedException,
            ExecutionException, TimeoutException {
        return null;
    }
    
    private Runnable getTask(){
        Runnable r = null;
        try {
            r = rQueue.poll(keepAliveTime, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return r;
    }
    
    private void addWorker(){
        Worker w = new Worker();
        w.start();
        lock.lock();
        workers.add(w);
        lock.unlock();
    }
    
    private void removeWorker(Worker w){
        lock.lock();
        workers.remove(w);
        lock.unlock();
    }
    
    class Worker extends Thread{
        
        private AtomicBoolean isAlive = new AtomicBoolean(true);
        
        private Runnable task;
        
        
        @Override
        public void run() {
            while(isAlive.get()){
                //阻塞一定时间,超时则回收该线程
                task = getTask();
                if(task != null){
                    task.run();
                }else{
                    isAlive.set(false);
                    
                }
                task = null;
            }
            System.out.println("remove worker");
            removeWorker(this);
        }
        
    }
    
    
}

测试代码：

package learnConcurrent;


public class ThreadPoolTest {
    static int taskNo = 0;
    public static void main(String[] args) throws InterruptedException {
        MyThreadPool pool = new MyThreadPool(2, 5);
        
        for(int i=0; i< 50; i++){
            Task task = new Task(taskNo++);
            pool.execute(task);
            Thread.sleep((int)(Math.random() * 1000));
        }
        
    }
    
}

class Task implements Runnable{
    String str;
    public Task(int taskNo){
        str = "TaskNo:" + taskNo;
    }
    @Override
    public void run() {
        System.out.println(str + " start work ");
        //DO SOMETHING
        try {
            Thread.sleep((int)(Math.random() * 1000));
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        
        System.out.println(str + " done ");
    }
    
}

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

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