线程池基本知识

野性酷女 2024-03-22 06:13 80阅读 0赞

## 线程池基本知识 ##

### 1、创建线程的方式 ###

#### 1.1、使用 继承Thread ####

#### 1.2、实现Runnable接口 ####

#### 1.3、实现Callable接口 ####

#### 1.4、线程池方式(Java提供了构建线程池的方式) ####

*  Java提供了Executors可以去创建(规范中不允许使用这种方式创建线程池,这种方式对线程的控制力度较低)
 *  手动创建线程池

### 2、熟悉线程池 ###

#### 2.1、线程池执行流程 ####

<table> 
 <thead> 
  <tr> 
   <th align="center">线程池执行流程</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center"><img src="https://img-blog.csdnimg.cn/img_convert/ea43c37ee05ed92403c0d7099ef97403.png" alt="线程池执行过程"></td> 
  </tr> 
 </tbody> 
</table>

#### 2.3、 线程池创建参数 ####

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.acc = System.getSecurityManager() == null ?
                    null :
                    AccessController.getContext();
            this.corePoolSize = corePoolSize;
            this.maximumPoolSize = maximumPoolSize;
            this.workQueue = workQueue;
            this.keepAliveTime = unit.toNanos(keepAliveTime);
            this.threadFactory = threadFactory;
            this.handler = handler;
        }

#### 2.3、 了解线程池的常量 ####

// 是一个int类型的数值, 表达两个意思:1、声明当前线程池的状态; 2、声明线程池中的线程数
        // 高3位是线程的状态   低29位是线程个数
        private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
        private static final int COUNT_BITS = Integer.SIZE - 3; // 获取线程
        private static final int CAPACITY   = (1 << COUNT_BITS) - 1; // 现层的容量
    
        // runState is stored in the high-order bits
        private static final int RUNNING    = -1 << COUNT_BITS; // 运行状态 高3位是 111
        private static final int SHUTDOWN   =  0 << COUNT_BITS; // 停止状态  000
        private static final int STOP       =  1 << COUNT_BITS;// 终止  001
        private static final int TIDYING    =  2 << COUNT_BITS; // 正在停止  010
        private static final int TERMINATED =  3 << COUNT_BITS; // 结束 011
    
        // 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; } //

#### 2.4、创建线程方法的解析 ####

public void execute(Runnable command) {
        
            // 线程不能为空
            if (command == null)
                throw new NullPointerException();
           // 获取 线程状态  和  线程数量
            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);
                // 获取工作线程状态是否为0
                else if (workerCountOf(recheck) == 0)
                    // 阻塞队列有任务,但是没有工作线程,添加一个任务为空的工作线程处理阻塞队列中的任务
                    addWorker(null, false);
            }
            // 创建非核心线程,处理阻塞队列中的任务
            else if (!addWorker(command, false))
                // 执行拒绝策略
                reject(command);
        }

#### 2.5、创建工作任务解析 ####

private boolean addWorker(Runnable firstTask, boolean core) {
        
      // 跳出嵌套循环的标记点      
      retry:
      // 死循环  给工作线程数标识 + 1
            for (;;) {
        
              // 获取ctl(32位)
                int c = ctl.get();
              // 获取线程池状态
                int rs = runStateOf(c);
    
                // Check if queue empty only if necessary.
              // 除了RUNNING都有可能
                if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                       firstTask == null &&
                       ! workQueue.isEmpty()))
                  // 1、rs == SHUTDOWN ;若不死SHUTDOWN就代表更高的状态,这时,不需要添加线程处理任务
                  // 2、任务为空  若果任务为null 且线程不是RUNNING,就不需要处理
                  // 3、阻塞队列不为null  如果阻塞队列为null,返回false,外侧取反,获取true,就不需要处理了
                  
                  // 构建工作线程失败
                    return false;
    
                for (;;) {
        
                  // 获取工作线程数量
                    int wc = workerCountOf(c);
                  	//     线程池的最大线程容量  如果当前线程已经大于线程池最大容量,不去创建了
                    if (wc >= CAPACITY ||
                        //   创建的是否为核心线程 ? 核心线程数量 : 最大线程数 判断工作线程是否超过核心线程或者最大线程数
                        wc >= (core ? corePoolSize : maximumPoolSize))
                      // 构建工作线程失败
                        return false;
                  // 将工作线程数+1 采用CAS的方式
                    if (compareAndIncrementWorkerCount(c))
                      // 成功推出到外层循环
                        break retry;
                  // 重新获取ctl
                    c = ctl.get();  // Re-read ctl
                  //  判断线程池状态是否与之前的一致, 如果一致重新执行内部循环
                    if (runStateOf(c) != rs)
                      // 若状态变化 结束这次外层循环,开始下次的外层循环
                        continue retry;
                    // else CAS failed due to workerCount change; retry inner loop
                }
            }
    				// 下面开始创建工作线程
            // worker开始
            boolean workerStarted = false;
      			// worker 添加
            boolean workerAdded = false;
      			// worker 就是工作线程
            Worker w = null;
            try {
        
              // 创建worker, 将任务传入
                w = new Worker(firstTask);
              // 从worker中获取线程t
                final Thread t = w.thread;
              // 线程不为null
                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());
    										// 线程状态为RUNNING
                        if (rs < SHUTDOWN ||
                            // 线程状态为SHUTDOWN。 或者 当前任务为null == 创建空线程任务处理阻塞队列中的任务 
                            (rs == SHUTDOWN && firstTask == null)) {
        
                          // 线程是否为运行状态
                            if (t.isAlive()) // precheck that t is startable
                                throw new IllegalThreadStateException();
                          // 开始将工作线程添加到集合中 HashSet<WOrker> 
                            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;
        }

#### 2.6、Worker 封装 ####

private final class Worker
            extends AbstractQueuedSynchronizer
            implements Runnable
        {
        
            /**
             * This class will never be serialized, but we provide a
             * serialVersionUID to suppress a javac warning.
             */
            private static final long serialVersionUID = 6138294804551838833L;
    
            /** Thread this worker is running in.  Null if factory fails. */
            final Thread thread;
            /** Initial task to run.  Possibly null. */
            Runnable firstTask;
            /** Per-thread task counter */
            volatile long completedTasks;
    
            /**
             * Creates with given first task and thread from ThreadFactory.
             * @param firstTask the first task (null if none)
             */
          // 获取到任务firstTask
            Worker(Runnable firstTask) {
        
                setState(-1); // inhibit interrupts until runWorker
                this.firstTask = firstTask;
              // 线程工厂创建一个线程,传入的参数时worker,调用worker的run方法   
                this.thread = getThreadFactory().newThread(this);
            }
    
          // 重写run方法
            public void run() {
        
                runWorker(this);
            }
    
            protected boolean isHeldExclusively() {
        
                return getState() != 0;
            }
    
            protected boolean tryAcquire(int unused) {
        
                if (compareAndSetState(0, 1)) {
        
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
                return false;
            }
    
            protected boolean tryRelease(int unused) {
        
                setExclusiveOwnerThread(null);
                setState(0);
                return true;
            }
    
            public void lock()        {
         acquire(1); }
            public boolean tryLock()  {
         return tryAcquire(1); }
            public void unlock()      {
         release(1); }
            public boolean isLocked() {
         return isHeldExclusively(); }
    
            void interruptIfStarted() {
        
                Thread t;
                if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
        
                    try {
        
                        t.interrupt();
                    } catch (SecurityException ignore) {
        
                    }
                }
            }
        }

#### 2.7、runWorker方法 ####

// 传入了 worker
        final void runWorker(Worker w) {
        
          // 获取当前线程
            Thread wt = Thread.currentThread();
          // 拿到任务
            Runnable task = w.firstTask;
          // 涉及到AQS
            w.firstTask = null;
          // 涉及到AQS
            w.unlock(); // allow interrupts
          // 标识为true
            boolean completedAbruptly = true;
            try {
        
              // 任务不为空,执行任务。  任务为空,通过getTask从阻塞队列中获取任务
                while (task != null || (task = getTask()) != null) {
        
                  // 获取锁,避免SHUTDOWN,终止当前任务
                    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
                  // 获取当前状态,是否大于等于STOP 
                    if ((runStateAtLeast(ctl.get(), STOP) ||
                         (Thread.interrupted() && // 获取状态
                          runStateAtLeast(ctl.get(), STOP))) &&
                        !wt.isInterrupted())
                      // 停止线程
                        wt.interrupt();
                    try {
        
                      // 执行任务前
                        beforeExecute(wt, task);
                        Throwable thrown = null;
                        try {
        
                          // Runnable.run(),开始执行任务
                            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);
            }
        }

#### 2.7、更多学习 ####

getTask、processWorkerExit、signal、await