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;
    }

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);//添加到队列失败或已满，做拒接任务处理策略
        }
        //若阻塞队列失败或已满；这里新建一个线程并启动做应急处理（这里就是用到了maximumPoolSize参数）
        else if (!addIfUnderMaximumPoolSize(command))
            reject(command); // 若线程池的线程超过了maximumPoolSize；就做拒绝处理任务策略
    }
}

private boolean addIfUnderCorePoolSize(Runnable firstTask) {
        Thread t = null;
        final ReentrantLock mainLock = this.mainLock;//获取当前线程池的锁
        mainLock.lock();//加锁
        try {
            /*
            这里线程池线程大小还需要判断一次；前面的判断过程中并没有加锁，因此可能在execute方法判断的时候poolSize小于corePoolSize，而判断完之后，在其他线程中又向线程池提交了任务，就可能导致poolSize不小于corePoolSize了，所以需要在这个地方继续判断
            */
            if (poolSize < corePoolSize && runState == RUNNING)
                t = addThread(firstTask);//新建线程
        } finally {
            mainLock.unlock();
        }
        if (t == null)
            return false;
        t.start();//若创建线程超过，就启动线程池的线程
        return true;
    }
    private Thread addThread(Runnable firstTask) {
        Worker w = new Worker(firstTask);//worker：ThreadPoolExecutor的内部类；
        Thread t = threadFactory.newThread(w);//使用线程工厂创建一个线程
        if (t != null) {
            w.thread = t;
            workers.add(w);//保存线程池正在运行的线程
            int nt = ++poolSize;//线程池的线程数加1
            if (nt > largestPoolSize)
                largestPoolSize = nt;
        }
        return t;
    }

public void run() {
    try {
        Runnable task = firstTask;//该线程的第一个任务，执行完后就从阻塞队列取任务执行
        firstTask = null;
        while (task != null || (task = getTask()) != null) {//getTask()从队列去任务执行
            runTask(task);//线程执行任务
            task = null;
        }
    } finally {
        workerDone(this);//若任务全部执行完，就开始尝试去停止线程池；这部分代码就不再追踪下去，有兴趣的读者可以自己打开源码分析，不必害怕，学习大神们的编码方式，看源码能让你学习到很多
    }
}
 private void runTask(Runnable task) {
    final ReentrantLock runLock = this.runLock;
    runLock.lock();
    try {
         //多次检查线程池有没有关闭
        if (runState < STOP &&
            Thread.interrupted() &&
            runState >= STOP)
            thread.interrupt();
            
        boolean ran = false;
        //这里就可以继承ThreadPoolExecutor,并覆盖beforeExecute(...)该方法，来做一些执行任务之前的统计工作或者用来保存正在执行的任务
        beforeExecute(thread, task);
        try {
            task.run();
            ran = true;
            //这里就可以继承ThreadPoolExecutor,并覆盖beforeExecute(...)该方法，来做一些执行任务完成之后的统计工作或者用来保存正在执行的任务
            afterExecute(task, null);
            ++completedTasks;//统计总共执行的任务数
        } catch (RuntimeException ex) {
            if (!ran)
                afterExecute(task, ex);
            throw ex;
        }
    } finally {
        runLock.unlock();
    }
}

public class MyExtendThreadPoolExecutor extends ThreadPoolExecutor{
    public static Logger logger=LoggerFactory.getLogger(MyExtendThreadPoolExecutor.class);
    /**
     * 记录运行中任务
     */
    private LinkedBlockingQueue<Runnable> workBlockingQueue=new  LinkedBlockingQueue<Runnable>();
    
    public MyExtendThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        workBlockingQueue.add((GtdataBreakpointResumeDownloadThread)r);//保存在运行的任务
        logger.info("Before the task execution");
    }
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        workBlockingQueue.remove((GtdataBreakpointResumeDownloadThread)r);//移除关闭的任务
        logger.info("After the task execution");
    }
    /**
     * 
    * Description: 正在运行的任务
    * @return LinkedBlockingQueue<Runnable><br>
    * @author lishun 
     */
    public LinkedBlockingQueue<Runnable> getWorkBlockingQueue() {
        return workBlockingQueue;
    }
}

public void addToThreadPool(DownloadRecord downloadRecord){
    BlockingQueue<Runnable> waitThreadQueue = pool.getQueue();//Returns the task queue 
    LinkedBlockingQueue<Runnable> workThreadQueue =pool.getWorkBlockingQueue();//Returns the running work
    GtdataBreakpointResumeDownloadThread downloadThread = 
            new GtdataBreakpointResumeDownloadThread(downloadRecord);//需要执行的任务线程
    
    if (!waitThreadQueue.contains(downloadThread)&&!workThreadQueue.contains(downloadThread)) {//判断任务是否存在正在运行的线程或存在阻塞队列，不存在的就加入线程池（这里的比较要重写equals()）
        Timestamp recordtime = new Timestamp(System.currentTimeMillis());
        logger.info("a_workThread:recordId="+downloadRecord.getId()+",name="+downloadRecord.getName()+" add to  workThreadQueue");
        downloadThread.setName("th_"+downloadRecord.getName());
        pool.execute(downloadThread);//添加到线程池
    }else{
        logger.info("i_workThread:recordId="+downloadRecord.getId()+",name="+downloadRecord.getName()+" in  waitThreadQueue or workThreadQueue");
    }
}