Android 线程池（一） FixedThreadPool、CachedThreadPool、SingleThreadPool 分析

ゝ一世哀愁。 2022-01-25 16:59 544阅读 0赞

### 目录 ###

*  线程池代码分析
 *   *  一、ThreadPoolExecutor
     *   *  1、构造方法
         *  2、测试代码
         *  3、源码分析
     *  二、FixedThreadPool 策略
     *   *  1、构造方法
         *  2 、源码分析
     *  三 、CachedThreadPool 策略
     *   *  1、构造方法
         *  2、源码分析
     *  四 、SingleThreadPool 策略
     *   *  1、构造方法
         *  2、源码分析

# 线程池代码分析 #

## 一、ThreadPoolExecutor ##

### 1、构造方法 ###

public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue)

*  corePoolSize 核心线程数，不会被回收，除非设置了allowcorethreadTimeout。
 *  maximumPoolSize 最大线程数，除核心线程外的线程会被回收
 *  keepAliveTime 当线程数大于核心时，这是非核心线程在终止前，等待新任务的最长时间。
 *  unit 时间单位
 *  workQueue 未执行的任务队列，如果超过了最大线程数的执行任务，都会放到这里。放不下会报 java.util.concurrent.RejectedExecutionException 异常。

### 2、测试代码 ###

*  任务代码

public class TestRunnable implements Runnable {
        private int index;
    
        public TestRunnable(int index) {
            this.index = index;
        }
    
        @Override
        public void run() {
            try {
                Thread.sleep(5000);
                Log.i("qinxue", "current Thread: " + Thread.currentThread().getName() + " index: " + index);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

*  执行代码

final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(3, 5, 1,
                    TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(10));//核心数3 最大线程个数5，空闲存活时间1，时间单位，工作队列
            ArrayList<TestRunnable> list = new ArrayList<>();
            for (int i = 0; i < 10; i++) {
                list.add(new TestRunnable(i)); //初始化任务队列
            }
    
            for (Runnable runnable : list) {
                threadPoolExecutor.execute(runnable); //提交给线程池执行
            }

*  执行结果  
    a.在提交的任务队列容量能够放的下的情况下，仅由核心线程执行，就那三个核心线程，不回收。  
    b.在提交任务队列容量放不下的情况下，会新建线程来执行，线程总数不超过最大线程数，剩余未执行的任务放队列，如果容量不够抛异常。如果这些新建线程空闲超过 keepAliveTime 就被销毁回收掉了。

### 3、源码分析 ###

*  常量分析

<table> 
 <thead> 
  <tr> 
   <th align="left">常量</th> 
   <th align="left">状态（高三位）</th> 
   <th align="left">计数 (低29位）</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">CAPACITY</td> 
   <td align="left">000</td> 
   <td align="left">1 1111 1111 1111 1111 1111 1111 1111</td> 
  </tr> 
  <tr> 
   <td align="left">RUNNING</td> 
   <td align="left">111</td> 
   <td align="left">0 0000 0000 0000 0000 0000 0000 0000</td> 
  </tr> 
  <tr> 
   <td align="left">SHUTDOWN</td> 
   <td align="left">000</td> 
   <td align="left">0 0000 0000 0000 0000 0000 0000 0000</td> 
  </tr> 
  <tr> 
   <td align="left">STOP</td> 
   <td align="left">001</td> 
   <td align="left">0 0000 0000 0000 0000 0000 0000 0000</td> 
  </tr> 
  <tr> 
   <td align="left">TIDYING</td> 
   <td align="left">010</td> 
   <td align="left">0 0000 0000 0000 0000 0000 0000 0000</td> 
  </tr> 
  <tr> 
   <td align="left">TERMINATED</td> 
   <td align="left">101</td> 
   <td align="left">0 0000 0000 0000 0000 0000 0000 0000</td> 
  </tr> 
 </tbody> 
</table>

TERMINATED>TIDYING>STOP>SHUTDOWN>RUNNING

*  变量分析

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    //原子操作的整形，初始状态 RUNNING 计数 0

*  基本方法

// Packing and unpacking ctl
        private static int runStateOf(int c)     { return c & ~CAPACITY; } 	//获取状态
        private static int workerCountOf(int c)  { return c & CAPACITY; }	//获取计数
        private static int ctlOf(int rs, int wc) { return rs | wc; }

*  核心方法

public void execute(Runnable command) {
            if (command == null)
                throw new NullPointerException();
            /*
             * Proceed in 3 steps:
             *
             * 1. If fewer than corePoolSize threads are running, try to
             * start a new thread with the given command as its first
             * task.  The call to addWorker atomically checks runState and
             * workerCount, and so prevents false alarms that would add
             * threads when it shouldn't, by returning false.
             *
             * 2. If a task can be successfully queued, then we still need
             * to double-check whether we should have added a thread
             * (because existing ones died since last checking) or that
             * the pool shut down since entry into this method. So we
             * recheck state and if necessary roll back the enqueuing if
             * stopped, or start a new thread if there are none.
             *
             * 3. If we cannot queue task, then we try to add a new
             * thread.  If it fails, we know we are shut down or saturated
             * and so reject the task.
             */
            int c = ctl.get();
            if (workerCountOf(c) < corePoolSize) { // 如果当前计数<核心线程数
                if (addWorker(command, true))		// 加入核心线程执行
                    return;											//成功就return了，后面不执行
                c = ctl.get();
            }
            if (isRunning(c) && workQueue.offer(command)) {	//如果加入队列成功，
                int recheck = ctl.get();
                if (! isRunning(recheck) && remove(command))	 //在队列中移除
                    reject(command);
                else if (workerCountOf(recheck) == 0)
                    addWorker(null, false);
            }
            else if (!addWorker(command, false))	//加入队列不成功，加入非核心线程执行
                reject(command);//在非核心线程执行失败抛异常，
        }

注释中也解释了代码  
1、如果当前计数小于核心线程数，新建线程并计数+1。  
2、如果加入队列成功，要在检测一次状态，如果不是运行状态出队，并抛出异常。  
3、如果加入队列失败，启动新的线程执行，如果失败直接抛出异常。

*  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()))	 //小于SHUTDOWN就只用RUNNING，不是RUNNNIG   返回失败
                    return false;
    
                for (;;) {
                    int wc = workerCountOf(c);
                    if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))		//超出最大线程数 返回失败
                        return false;
                    if (compareAndIncrementWorkerCount(c))					//未超出时计数+1
                        break retry;																//跳出外部循环
                    c = ctl.get();  // Re-read ctl
                    if (runStateOf(c) != rs)												//CAS失败，状态改变，外循环重试
                        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);						//包装成Worker然后执行，后面是执行代码
                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;
        }

*  实际运行

final void runWorker(Worker w) {
            Thread wt = Thread.currentThread();
            Runnable task = w.firstTask;
            w.firstTask = null;
            w.unlock(); // allow interrupts
            boolean completedAbruptly = true;
            try {
            //线程会不断从队列中获取任务来执行，重复利用线程
            //如果为null则退出while线程也就退出了：这里注意的一点是核心线程就是在这里阻塞没有被销毁的
                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);
            }
        }

*  getTask

private Runnable getTask() {
            boolean timedOut = false; // Did the last poll() time out?
    
            for (;;) {
                int c = ctl.get();
                int rs = runStateOf(c);
    
                // Check if queue empty only if necessary.
                if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
                    decrementWorkerCount();
                    return null;
                }
    
                int wc = workerCountOf(c);
    
                // Are workers subject to culling?
                boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    			
    			//即使超时，但wc>corePoolSize 不会执行 下面的线程销毁和-1，会循环去取，核心线程就不会被销毁
                if ((wc > maximumPoolSize || (timed && timedOut))
                    && (wc > 1 || workQueue.isEmpty())) {
                    if (compareAndDecrementWorkerCount(c))	//返回null，线程程会被销毁因此-1
                        return null;
                    continue;
                }
    
                try {
                    Runnable r = timed ?
                        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) ://最多等待keepAliveTime
                        workQueue.take();			
                    if (r != null)
                        return r;
                    timedOut = true;
                } catch (InterruptedException retry) {
                    timedOut = false;
                }
            }
        }

*  shutDown

public void shutdown() {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                checkShutdownAccess();		//是否可以关闭
                advanceRunState(SHUTDOWN);		//切换状态
                interruptIdleWorkers();					//中断任务
                onShutdown(); // hook for ScheduledThreadPoolExecutor 
            } finally {
                mainLock.unlock();
            }
            tryTerminate();						//结束线程池
        }
    
            public List<Runnable> shutdownNow() {
            List<Runnable> tasks;
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                checkShutdownAccess();
                advanceRunState(STOP);
                interruptWorkers();
                tasks = drainQueue();  //清空未执行的任务并返回
            } finally {
                mainLock.unlock();
            }
            tryTerminate();
            return tasks;
        }

当调用shutdown方法时，线程池将不会再接收新的任务，然后将先前放在队列中的任务执行完成。

当调用shutdownNow方法时立即停止所有的执行任务，并将队列中的任务返回。

*  tryTerminate

final void tryTerminate() {
            for (;;) {
                int c = ctl.get();
                if (isRunning(c) ||
                    runStateAtLeast(c, TIDYING) ||
                    (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
                    return;			//状态不正确或者状态正确SHUTDOWN但是队列不为null重试
                if (workerCountOf(c) != 0) { // Eligible to terminate			// 线程中任务计数不为0 重试
                    interruptIdleWorkers(ONLY_ONE);
                    return;
                }
    
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {	// 设置为 TIDYING（计数和队列里都没有任务了）
                        try {
                            terminated();
                        } finally {
                            ctl.set(ctlOf(TERMINATED, 0));					//设置为 TERMINATED
                            termination.signalAll();
                        }
                        return;
                    }
                } finally {
                    mainLock.unlock();
                }
                // else retry on failed CAS
            }
        }

以上大体了解了整个流程。

## 二、FixedThreadPool 策略 ##

### 1、构造方法 ###

public static ExecutorService newFixedThreadPool(int nThreads) {
            return new ThreadPoolExecutor(nThreads, nThreads,
                                          0L, TimeUnit.MILLISECONDS,
                                          new LinkedBlockingQueue<Runnable>());
        }

### 2 、源码分析 ###

a.这里核心线程数和最大线程数一致  
b.不等待任务  
c.任务队列不设置限制  
适用：大量短生命周期的异步任务时（many short-lived asynchronous task）

## 三 、CachedThreadPool 策略 ##

### 1、构造方法 ###

public static ExecutorService newCachedThreadPool() {
            return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                          60L, TimeUnit.SECONDS,
                                          new SynchronousQueue<Runnable>());
        }

### 2、源码分析 ###

a、核心线程为0则不存在常驻线程  
b、最大线程相当于不设置限制，随时创建和使用  
c、等待60s，等待时间的增加，提高了线程的重复利用率。60s之内提交线程就会重复利用  
d、SynchronousQueue 内部没有容量，但是由于一个插入操作总是对应一个移除操作，反过来同样需要满足。那么一个元素就不会再SynchronousQueue 里面长时间停留，一旦有了插入线程和移除线程，元素很快就从插入线程移交给移除线程。后面再看下这个。  
适用：稀疏提交任务，任务声明周期较短的任务，提交立马被消费的情况。

## 四 、SingleThreadPool 策略 ##

### 1、构造方法 ###

public static ExecutorService newSingleThreadExecutor() {
            return new FinalizableDelegatedExecutorService
                (new ThreadPoolExecutor(1, 1,
                                        0L, TimeUnit.MILLISECONDS,
                                        new LinkedBlockingQueue<Runnable>()));
        }

### 2、源码分析 ###

a、核心线程和最大线程数都是1，只有个常驻线程循环适用  
b、等待时间不等待。  
适用：SingleThreadExecutor 适用于在逻辑上需要单线程处理任务的场景

ScheduledThreadPoolExecutor、WorkStealingPool 的实现与以上策略不同，后面分析。