线程池的实现原理，绝对简单易懂

刺骨的言语ヽ痛彻心扉 2023-06-16 04:56 31阅读 0赞

### 文章目录 ###

*  线程池实现原理
 *   *  1.线程池简介
     *  2.Java中提供的线程池
     *  3.ThreadPoolExecutor实现原理

# 线程池实现原理 #

## 1.线程池简介 ##

为什么要构建线程池？

1.复用已有的资源（减少cpu时间片的切换）

2.控制现场资源总数

优势：

1.限流->控制线程数量

2.降低频繁创建和销毁线程。（对于任务的响应速度更快，可以直接从线程池中取，不需要创建线程）

## 2.Java中提供的线程池 ##

Executors,线程池工具类 。

//主要都是ThreadPoolExecutor 构造的线程池
    //ThreadPoolExecutor的构造方法
    public ThreadPoolExecutor(int corePoolSize,//核心线程数
                                  int maximumPoolSize,最大线程数
                                  long keepAliveTime,//保持连接时间
                                  TimeUnit unit,//保持连接时间单位
                                  BlockingQueue<Runnable> workQueue,//阻塞队列
                                  ThreadFactory threadFactory,//线程工厂
                                  RejectedExecutionHandler handler) //拒绝策略

newFixedThreadPool()

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

newCachedThreadPool

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

newSingleThreadScheduledExecutor

public static ScheduledExecutorService newSingleThreadScheduledExecutor() { 
            return new DelegatedScheduledExecutorService
                (new ScheduledThreadPoolExecutor(1));
        }

newScheduledThreadPool

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) { 
            return new ScheduledThreadPoolExecutor(corePoolSize);
        }

public ScheduledThreadPoolExecutor(int corePoolSize) { 
            super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
                  new DelayedWorkQueue());
        }

newWorkStealingPool

public static ExecutorService newWorkStealingPool(int parallelism) { 
            return new ForkJoinPool
                (parallelism,
                 ForkJoinPool.defaultForkJoinWorkerThreadFactory,
                 null, true);
        }

//ForkJoinPool构造方法。
    public ForkJoinPool(int parallelism,
                            ForkJoinWorkerThreadFactory factory,
                            UncaughtExceptionHandler handler,
                            boolean asyncMode) { 
            this(checkParallelism(parallelism),
                 checkFactory(factory),
                 handler,
                 asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
                 "ForkJoinPool-" + nextPoolId() + "-worker-");
            checkPermission();
        }

问题1:猜想keepAliveTime如何去监控线程进行回收？下面我们看看线程池是如何实现的。  
1.有一个是否回收核心线程的开关(allowCoreThreadTimeOut)  
2.当前线程数是否大于核心线程数（wc > corePoolSize）  
请参考getTask()方法，只要返回空，线程则会进行回收。

## 3.ThreadPoolExecutor实现原理 ##

execute方法实现如下：

//线程默认运行状态是 RUNNING
    	private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
        private static final int COUNT_BITS = Integer.SIZE - 3;
    	//表示低29位全是1，高3位为0
        private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    
        // runState is stored in the high-order bits
    	//高3位存储线程运行状态
        private static final int RUNNING    = -1 << COUNT_BITS;
        private static final int SHUTDOWN   =  0 << COUNT_BITS;
        private static final int STOP       =  1 << COUNT_BITS;
        private static final int TIDYING    =  2 << COUNT_BITS;
        private static final int TERMINATED =  3 << COUNT_BITS;
    
        // Packing and unpacking ctl
    	//获取高三位，表示获取线程运行状态
        private static int runStateOf(int c)     {  return c & ~CAPACITY; }
    	//获取低29位，表示获取线程数量
        private static int workerCountOf(int c)  {  return c & CAPACITY; }
        private static int ctlOf(int rs, int wc) {  return rs | wc; }
    //ThreadPoolExecutor.execute实现源码
    public void execute(Runnable command) { 
            if (command == null)
                throw new NullPointerException();
            /* * Proceed in 3 steps://分为3步 * * 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. */
      			//ctl默认状态为运行状态 
            int c = ctl.get();
            if (workerCountOf(c) < corePoolSize) { 
                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);
                else if (workerCountOf(recheck) == 0)
                   addWorker(null, false);
            }
            else if (!addWorker(command, false))
                reject(command);
        }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70]

addWorker：

//firstTask 传递的是.execute(Runnable r) 中的r,core 表示是否创建一个核心线程数
    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;
                    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;
        }

主要做两件事：

//主要增加工作线程数
    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;
                  //最主要做的事情 CAS给ctl加1，增加一个工作线程数
                    if (compareAndIncrementWorkerCount(c))
                        break retry;
                  //重试获取，再一次读区ctl
                    c = ctl.get();  // Re-read ctl
                  //如果当前状态不是之前的状态，则继续retry
                    if (runStateOf(c) != rs)
                        continue retry;
                    // else CAS failed due to workerCount change; retry inner loop
                }
            }

//主要创建一个Worker线程，并添加到队列(真正意义的构建线程)
    				boolean workerStarted = false;
            boolean workerAdded = false;
            Worker w = null;
            try { 
                w = new Worker(firstTask);
              	/** *Worker(Runnable firstTask) { * 初始化state为-1。 * setState(-1); // inhibit interrupts until runWorker * this.firstTask = firstTask; * 把当前Worker对象赋值为Worker中的thread * this.thread = getThreadFactory().newThread(this); *} * */
                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());
    										//当前状态为Runnable 或者（状态为SHUTDOWN，firstTask为null）
                        if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) { 
                            if (t.isAlive()) // precheck that t is startable
                                throw new IllegalThreadStateException();
                    //private final HashSet<Worker> workers = new HashSet<Worker>();
                            //把当前worker添加到workers Set中
                          	workers.add(w);
                            int s = workers.size();
                            if (s > largestPoolSize)
                                largestPoolSize = s;
                            workerAdded = true;
                        }
                    } finally { 
                        mainLock.unlock();
                    }
                    if (workerAdded) { 
                      	/** * public void run() { * runWorker(this); * } * */
                        t.start();
                        workerStarted = true;
                    }
                }
            } finally { 
                if (! workerStarted)
              //如果执行失败，需要回滚
              /** * private void addWorkerFailed(Worker w) { * final ReentrantLock mainLock = this.mainLock; * mainLock.lock(); * try { * if (w != null) * workers.remove(w); * //把数量减1 CAS实现 * decrementWorkerCount(); * tryTerminate(); * } finally { * mainLock.unlock(); * } * } */
                    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 { 方法后面讲
              	//getTask是从阻塞队列里获取，已经不是直接交给核心线程了。
                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 { 
                          	//问题2.为什么要调用run方法呢？
                          	//task这时候已经存在线程池中了，所以没有必要.start方法，再创建一个线程去执行了
                            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);
            }
        }

Worker：实现了Runnable，并且继承了AQS

问题3:Worker为什么要实现AQS呢？为什么不用ReentrantLock实现呢？  
![在这里插入图片描述][20191124170422594.png]  
我们看一下Worker里面做了什么？

Worker(Runnable firstTask) { 
      					//设置同步状态，
                setState(-1); // inhibit interrupts until runWorker
                this.firstTask = firstTask;
                this.thread = getThreadFactory().newThread(this);
            }

protected boolean tryAcquire(int unused) { 
      					//1是表示获取锁，0表示释放状态，-1是初始化状态
                if (compareAndSetState(0, 1)) { 
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
                return false;
            }

protected boolean tryRelease(int unused) { 
                setExclusiveOwnerThread(null);
                setState(0);
                return true;
            }

//上面的runWorker 为什么w要加锁呢？
    //ThreadPoolExecutor.shutdown方法
        public void shutdown() { 
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try { 
                checkShutdownAccess();
                advanceRunState(SHUTDOWN);
                interruptIdleWorkers();
                onShutdown(); // hook for ScheduledThreadPoolExecutor
            } finally { 
                mainLock.unlock();
            }
            tryTerminate();
        }

//interruptIdleWorkers方法，中断时会判断，线程是否还在运行。
      private void interruptIdleWorkers(boolean onlyOne) { 
              final ReentrantLock mainLock = this.mainLock;
              mainLock.lock();
              try { 
                  for (Worker w : workers) { 
                      Thread t = w.thread;			
                      if (!t.isInterrupted() && /*重点*/ w.tryLock()) { 
                          try { 
                              t.interrupt();
                          } catch (SecurityException ignore) { 
                          } finally { 
                              w.unlock();
                          }
                      }
                      if (onlyOne)
                          break;
                  }
              } finally { 
                  mainLock.unlock();
              }
          }

这就是为什么Worker要实现AQS的原因，为了shuatdown时不终止正在运行的任务

//getTask从workQueue获取执行的线程，非核心线程数是否被回收？
    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;
    
            if ((wc > maximumPoolSize || (timed && timedOut))
                && (wc > 1 || workQueue.isEmpty())) { 
                if (compareAndDecrementWorkerCount(c))
                    return null;
                continue;
            }
    
            try { 
                Runnable r = timed ?
                    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    workQueue.take();
                if (r != null)
                    return r;
                timedOut = true;
            } catch (InterruptedException retry) { 
                timedOut = false;
            }
        }
    }

执行流程：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70 1]

图例：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70 2]

队列的简单操作
    offer       添加一个元素并返回true      如果队列已满，则返回false
    poll        移除并返问队列头部的元素     如果队列为空，则返回null
    put         添加一个元素               如果队列满，则阻塞
    take        移除并返回队列头部的元素     如果队列为空，则阻塞

1.不建议使用Executors 创建线程

2.线程的执行情况->

IO密集型 CPU核心数的2倍

CPU密集型 CPU核心数+1

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70 3]

Executor框架最核心的接口是Executor，它表示任务的执行器。

Executor的子接口为ExecutorService。

ExecutorService有两大实现类：ThreadPoolExecutor和ScheduledThreadPoolExecutor。

注：本文为博主原创文章，转载请附上原文出处链接和本声明。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191124170723475.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1,size_16,color_FFFFFF,t_70
[20191124170422594.png]: https://img-blog.csdnimg.cn/20191124170422594.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191124170222777.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20191124170304638.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20191124170326554.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxNDMwNjY1,size_16,color_FFFFFF,t_70