Java线程池ThreadPoolExecutor

你的名字 2022-09-25 09:23 186阅读 0赞

### 为什么使用线程池 ###

如果每次使用线程都创建，每次创建和销毁的开销会很大，线程池主要用来解决线程生命周期开销问题和资源不足问题，也消除了线程创建所带来的延迟。

### 线程池分析 ###

先看下构造函数，其他构造函数都是调用此构造函数

public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  ThreadFactory threadFactory,
                                  RejectedExecutionHandler handler) {
    
    }

参数分析  
corePoolSize核心线程池的大小。  
maximumPoolSize最大允许的线程池大小。  
keepAliveTime，unit当线程池数大于corePoolSize时，当一个线程空闲时间超过此时间，将会结束。  
workQueue阻塞队列，用于存储还没有执行的任务。  
threadFactory用来创建线程。  
handler拒绝处理策略，当线程池和队列都满了，有一下几种策略

> ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。  
> ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。  
> ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）  
> ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

当向线程池提交任务时，将会执行execute()方法。

public void execute(Runnable command) {
     //如果为null，抛出空指针异常
            if (command == null)
                throw new NullPointerException();
    
            int c = ctl.get();
            //当运行的线程小于核心池大小时
            if (workerCountOf(c) < corePoolSize) {
                //添加任务，失败返回false
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }
            //线程池在运行态，尝试加入线程池
            if (isRunning(c) && workQueue.offer(command)) {
                int recheck = ctl.get();
                //如果线程池不在运行态，将任务移除
                if (! isRunning(recheck) && remove(command))
                //移除任务失败，执行拒绝策略
                    reject(command);
                    //检查一下当前线程池的缓存队列数量是否为0
                else if (workerCountOf(recheck) == 0)
                //是否需要新加一个线程
                    addWorker(null, false);
            }
            //再次添加任务，如果失败，就是线程池满了
            else if (!addWorker(command, false))
                reject(command);
        }

再来看看addWorker()方法，添加任务，并执行

private boolean addWorker(Runnable firstTask, boolean core) {
            retry:
            for (;;) {
                int c = ctl.get();
                int rs = runStateOf(c);
    
                // Check if queue empty only if necessary.
                if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                       firstTask == null &&
                       ! workQueue.isEmpty()))
                    return false;
    
                for (;;) {
                //获取线程数
                    int wc = workerCountOf(c);
                    if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                        return false;
                        //线程池未过限,将线程池计数器扩增1，跳出标签
                    if (compareAndIncrementWorkerCount(c))
                        break retry;
                    c = ctl.get();  // Re-read ctl
                    if (runStateOf(c) != rs)
                        continue retry;
                    // else CAS failed due to workerCount change; retry inner loop
                }
            }
    
            boolean workerStarted = false;
            boolean workerAdded = false;
            Worker w = null;
            try {
                w = new Worker(firstTask);
                final Thread t = w.thread;
                if (t != null) {
                    final ReentrantLock mainLock = this.mainLock;
                    mainLock.lock();
                    try {
                        // Recheck while holding lock.
                        // Back out on ThreadFactory failure or if
                        // shut down before lock acquired.
                        int rs = runStateOf(ctl.get());
    
                        if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) {
                            if (t.isAlive()) // precheck that t is startable
                                throw new IllegalThreadStateException();
                                //向运行线程池内加上该线程
                            workers.add(w);
                            int s = workers.size();
                            if (s > largestPoolSize)
                                largestPoolSize = s;
                            workerAdded = true;
                        }
                    } finally {
                        mainLock.unlock();
                    }
                    if (workerAdded) {
                        t.start();
                        workerStarted = true;
                    }
                }
            } finally {
                if (! workerStarted)
                //如果任务启动失败，回滚前面的操作中断线程池或者中断当前线程
                    addWorkerFailed(w);
            }
            return workerStarted;
        }

接下来看执行线程的方法runWorker()

final void runWorker(Worker w) {
            Thread wt = Thread.currentThread();
            Runnable task = w.firstTask;
            w.firstTask = null;
            w.unlock(); // allow interrupts
            boolean completedAbruptly = true;
            try {
            //循环获取任务和子任务
                while (task != null || (task = getTask()) != null) {
                    w.lock();
                    // If pool is stopping, ensure thread is interrupted;
                    // if not, ensure thread is not interrupted. This
                    // requires a recheck in second case to deal with
                    // shutdownNow race while clearing interrupt
                    if ((runStateAtLeast(ctl.get(), STOP) ||
                         (Thread.interrupted() &&
                          runStateAtLeast(ctl.get(), STOP))) &&
                        !wt.isInterrupted())
                        wt.interrupt();
                    try {
                    //扩展方法，提供子类实现
                        beforeExecute(wt, task);
                        Throwable thrown = null;
                        try {
                        //当前线程运行
                            task.run();
                        } catch (RuntimeException x) {
                            thrown = x; throw x;
                        } catch (Error x) {
                            thrown = x; throw x;
                        } catch (Throwable x) {
                            thrown = x; throw new Error(x);
                        } finally {
                            afterExecute(task, thrown);
                        }
                    } finally {
                        task = null;
                        w.completedTasks++;
                        w.unlock();
                    }
                }
                completedAbruptly = false;
            } finally {
                processWorkerExit(w, completedAbruptly);
            }
        }

线程池的主要方法就学习到这。

### 其他参考 ###

[http://blog.csdn.net/zclzhangcl/article/details/50772434][http_blog.csdn.net_zclzhangcl_article_details_50772434]

[http://blog.csdn.net/u013142781/article/details/51387749][http_blog.csdn.net_u013142781_article_details_51387749]

[http_blog.csdn.net_zclzhangcl_article_details_50772434]: http://blog.csdn.net/zclzhangcl/article/details/50772434
[http_blog.csdn.net_u013142781_article_details_51387749]: http://blog.csdn.net/u013142781/article/details/51387749