Java并发：线程池

ゞ浴缸里的玫瑰 2022-12-02 10:53 231阅读 0赞

### 线程目的 ###

1.  线程复用，线程的创建与销毁是很消耗资源的。
2.  防止出现OOM

### ThreadPoolExecutor ###

直接来看下构造方法，不过一般构造线程池会用Executors这个工具类，且常用的方法有三个，有兴趣可以去看看，就直接上个最长的构造器。

// 可自定义扩展线程池
    	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;
        }

构造器有七个参数，下面一一解答：

*  corPoolSize：核⼼线程数，线程池中有核心线程+非核心线程
 *  maxmunPoolSiza：线程池能创建的最⼤线程数量，核心线程数+非核心线程数
 *  keepAliveTime：非核心线程允许的空闲时间
 *  unit：时间对象
 *  workQueue：阻塞队列，这个队列维护的是等待执行的Runnable对象。
 *  ThreadFactory：线程⼯⼚
 *  handler：任务拒绝策略

常用的有四种处理策略：

*  AbortPolicy ：默认的拒绝策略，会丢弃任务并抛出RejectedExecutionException 异常
 *  CallerRunsPolicy ：提交任务的线程，自己去执行这个任务
 *  DiscardPolicy ：直接丢弃新来的任务，也没有任何异常抛出
 *  DiscardOldestPolicy ：丢弃最老的任务，然后将新任务加入到工作队列中

默认拒绝策略是 AbortPolicy 。

**关键属性：**  
ThreadPoolExecutor类中采用原子类成员变量ctl来同时记录线程池的状态和线程池中线程的数量，分别用最高3位和剩余29表示。

// ctl保存两个信息：任务数量和线程池状态，默认为RUNNING，3位高位表示运行状态，低位29位表示线程池中 活动线程的数量
    	private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
    	// 29
        private static final int COUNT_BITS = Integer.SIZE - 3;
    	// 线程最大个数 000111...11 3个0和29个1
        private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
        // 线程运行状态
        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;
        // 返回运行状态
        private static int runStateOf(int c)     {  return c & ~CAPACITY; }
    	// 返回线程池中活动线程的个数
        private static int workerCountOf(int c)  {  return c & CAPACITY; }
    	// 返回rs，也就是RUNNING
        private static int ctlOf(int rs, int wc) {  return rs | wc; }

**execute方法：** 下面就是执行任务的方法了

/** * 在未来的某个时候执行给定任务。该任务可以在一个新的线程或在现有池线程中执行。 如果任务不能执行提交， * 或者是因为该执行程序已关闭或因为它的容量已经达到，任务由当前处理RejectedExecutionHandler */
    	public void execute(Runnable command) { 
            // 报错
            if (command == null)
                throw new NullPointerException();
            //获取ctl ，运行状态+线程数量
            int c = ctl.get();
            //如果线程池中线程数量 < 核心线程数量，则创建新的线程来处理请求，即使其他线程是空闲的
            if (workerCountOf(c) < corePoolSize) { 
                //创建新线程执行任务
                if (addWorker(command, true))
                    return;
                c = ctl.get();
            }
            // 如果线程池中线程数量>=corePoolSize，且线程池处于RUNNING状态，且把提交的任务成
    		// 功放⼊阻塞队列中，就再次检查线程池的状态，
            if (isRunning(c) && workQueue.offer(command)) { 
                int recheck = ctl.get();
                // 1.如果线程池不是RUNNING状态，且成功从阻塞队列中删除任务，则该任务由当前
    		    // RejectedExecutionHandler处理。回滚排队
                if (! isRunning(recheck) && remove(command))
                    reject(command);
                // 2.如果线程池中运⾏的线程数量为0，则通过addWorker(null, false)尝试新建⼀个
                // 线程，新建线程对应的任务为null。
                else if (workerCountOf(recheck) == 0)
                    addWorker(null, false);
            }
            // 如果以上两种case不成⽴，即没能将任务成功放⼊阻塞队列中，会尝试创建非核心线程运行该任务，都失 // 败则说明任务队列满了，线程池也满了，该任务由当前 RejectedExecutionHandler 处理。
            else if (!addWorker(command, false))
                reject(command);
        }

**总结：**

*  如果Runnable为null，报错。
 *  如果线程池中的线程数量小于核心线程数，创建新的线程运行该任务。
 *  如果线程池中的线程数量大于等于核心线程数，加入到等待队列中；加入成功对线程池状态进行再次确认，线程池状态不为running时，进行回滚排队；如果线程池中的线程数量为0，则尝试创建一个新线程。
 *  如果没有办法给任务排队，可能是队列满了或者线程池关闭了，执行拒绝策略。  
    **流程：**  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMDA4NTM1_size_16_color_FFFFFF_t_70_pic_center]  
    **addWorker方法：**

//添加任务
        // 线程数线程数量小于核心线程数，addWorker(command, true)
    	// 线程数量大于等于核心线程数且添加任务到任务队列失败，addWorker(command, false)
        private boolean addWorker(Runnable firstTask, boolean core) { 
            retry:
            for (;;) { 
                // 获取state
                int c = ctl.get();
                // 获取线程池运行状态
                int rs = runStateOf(c);
              // 线程池状态 >= SHUTDOWN 时,不再接受新的任务，也就是线程池状态不为RUNNING,直接返回 false
              // 如果 rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty() 同样不接受新的任务,返回 false
                // 如果线程池状态为SHUTDOWN时，不接受新任务，会运行完阻塞队列里的任务
                // 线程状态为SHUTDOWN 且 任务为null 且任务等待队列不为空
                if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&firstTask == null &&! workQueue.isEmpty()))
                    return false;
    			// 
                for (;;) { 
                    // 获取线程池中的线程个数
                    int wc = workerCountOf(c);
                    // 如果wc >= CAPACITY，线程池中的线程数大于等于最大线程数，返回false
                    // wc >= (core ? corePoolSize : maximumPoolSize)
                    // core：true/false，创建的新线程为核心线程还是非核心线程
                    if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                        return false;
                    // cas修改workCount+1成功，则跳出循环
                    if (compareAndIncrementWorkerCount(c))
                        break retry;
                    //获取状态
                    c = ctl.get();  
                    // 线程池的状态有变化，则重试
                    if (runStateOf(c) != rs)
                        continue retry; 
                }
                
            }
            
    		// workCount增加成功之后的操作：
            boolean workerStarted = false;
            boolean workerAdded = false;
            Worker w = null;
            try { 
                // 创建一个work对象
                w = new Worker(firstTask);
                // 从worker对象获取线程对象
                final Thread t = w.thread;
                // 如果线程对象不为null
                if (t != null) { 
                    // 获取锁
                    // mainLock是ThreadPoolExecutor的成员变量
                    // private final ReentrantLock mainLock = new ReentrantLock();
                    final ReentrantLock mainLock = this.mainLock;
                    mainLock.lock();
                    // 添加任务到HashSet集合中
                    try { 
    					// 获取线程池状态
                        int rs = runStateOf(ctl.get());
    					// 如果rs < SHUTDOWN，这个就是线程池状态为RUNNING
                        // 如果rs == SHUTDOWN && firstTask == null，线程状态为SHUTDOWN
                        if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) { 
                            if (t.isAlive()) 
                                throw new IllegalThreadStateException();
                            // 将任务添加到线程池中
                            // workers是hashSet集合
                            // private final HashSet<Worker> workers = new HashSet<Worker>(); //添加work对象到HashSet集合中
                            workers.add(w);
                            // 获取集合的数量
                            int s = workers.size();
                            if (s > largestPoolSize)
                                largestPoolSize = s;
                            // 添加工人到集合成功
                            workerAdded = true;
                        }
                    } finally { 
                        mainLock.unlock();
                    }
                    
                    // 如果添加成功
                    if (workerAdded) { 
                        // 启动任务线程,开始执行任务
                        // 我们需要看看start()方法
                        t.start();
                        workerStarted = true;
                    }
                }
            } finally { 
                if (! workerStarted)
           // 如果任务线程启动失败调用 addWorkerFailed 
           // addWorkerFailed方法里面主要做了两件事:将该线程从线程池中移除;将 workerCount 的值减 1
                    addWorkerFailed(w);
            }
            return workerStarted;
        }

方法功能如其名，添加工作者，往哪里添加呢，向workers集合，workers集合的本质为HashSet，用来保存Worker对象的。

**Worker：**

线程池维护的是一个HashSet，一个由worker对象组成的HashSet。

private final HashSet<Worker> workers = new HashSet<Worker>();

worker 继承 AQS 主要是利用 AQS 独占锁机制，来标识线程是否空闲；另外， worker 还实现了 `Runnable` 接口，所以它本身就是一个线程任务，在构造方法中创建了一个线程，线程的任务就是自己 `this.thread = getThreadFactory().newThread(this);`

private final class Worker
            extends AbstractQueuedSynchronizer
            implements Runnable
        { 
            private static final long serialVersionUID = 6138294804551838833L;
            // 执行任务的线程
            final Thread thread;
            // 传入的任务
            Runnable firstTask;
            volatile long completedTasks;
            Worker(Runnable firstTask) { 
                // 设置AQS的state
                setState(-1);
                this.firstTask = firstTask;
                // this指的是worker对象
                this.thread = getThreadFactory().newThread(this);
            }
            public void run() { 
                runWorker(this);
            }
    		// ... 各种方法
        }

**runWork：**

final void runWorker(Worker w) { 
            // 获取当前线程
            Thread wt = Thread.currentThread();
            // 获取任务
            Runnable task = w.firstTask;
            w.firstTask = null;
            // 允许中断
            w.unlock(); 
            boolean completedAbruptly = true;
            try { 
                // 判断task是否为空，不空则直接执行
                // 如果为空，则调用getTask()方法从workQueue中取出新的task执行
                while (task != null || (task = getTask()) != null) { 
                    // 加锁
                    w.lock();
                    // 检查线程池状态为stop则需要中断该线程
                    if (( runStateAtLeast(ctl.get(), STOP) ||
                        ( Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP) ) )
                        && !wt.isInterrupted())
                        wt.interrupt();
                    
                    try { 
                        // 执行beforeExecute，自定义方法
                        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 方法，自定义任务
                            afterExecute(task, thrown);
                        }
                    } finally { 
                        // 将 task 设置为 null ,循环操作
                        task = null;
                        w.completedTasks++;
                        // 释放锁
                        w.unlock();
                    }
                }
                completedAbruptly = false;
            } finally { 
            	// 移除非核心线程
                processWorkerExit(w, completedAbruptly);
            }
            
        }

**getTask:**

private Runnable getTask() { 
            boolean timedOut = false; 
    
            for (;;) { 
                // 获取状态
                int c = ctl.get();
                // 获取运行状态
                int rs = runStateOf(c);
    		    // 如果线程池状态不为RUNNING 并且 rs >= STOP或者任务队列为空
                if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { 
                    decrementWorkerCount();
                    return null;
                }
    			// 获取线程池中的线程数
                int wc = workerCountOf(c);
    			// 定义是否需要线程过期
                boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    			// 如果运行线程数大于最大线程数,但是缓存队列已经空了,此时递减 worker 数量
                // 如果有设置允许线程超时或者线程数量超过了核心线程数量,并且线程在规定时间内没有 poll 到任务并且队列为空,此时也递减 worker 数量
                if ( (wc > maximumPoolSize || (timed && timedOut))
                    && (wc > 1 || workQueue.isEmpty())) { 
                    if (compareAndDecrementWorkerCount(c))
                        return null;
                    continue;
                }
    
                try { 
                    // 如果 timed 为 true ,会调用 workQueue 的 poll 方法
                    // 超时时间为 keepAliveTime ,如果超过 keepAliveTime 时长的话, poll 就会返回 null 
                    // 如果返回为 null ,在 runWorker 中 
                    // while (task != null || (task = getTask()) != null) 循环条件被打破,从而跳出循环,此时线程执行完毕
                    // 如果 timed 为 false ( allowCoreThreadTimeOut 为 false ,并且 wc > corePoolSize 为 false )
                    // 会调用 workQueue 的 take 方法阻塞到当前
                    // 当队列中有任务加入时,线程被唤醒, take 方法返回任务,开始执行
                    Runnable r = timed ?
                        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                    	// 挂起
                    	// condition条件队列的唤醒与挂起
                        workQueue.take();
                    if (r != null)
                        return r;
                    timedOut = true;
                } catch (InterruptedException retry) { 
                    timedOut = false;
                }
            }
    }

**要点：**

*  workQueue：先进先出的阻塞队列，用来保存任务
 *  workers：本质为HashSet集合，用来保存工作线程，也包括非核心线程。

线程复用重点就在在 runWorker 方法中的 while 循环，在 while 循环里面， worker 会不断调用 getTask 方法，而在 getTask 方法里，如果任务队列中没有了任务，会出现两种情况，①：如果线程为核心线程，则会因为调用 workQueue的take 方法而挂起线程，直到拿到任务然后返回 true ，；②：如果线程为非核心线程，也就是会过期的线程，它就会因为拿不到任而结束循环，再调用processWorkerExit(w, completedAbruptly)来移除自己。  
**流程：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMDA4NTM1_size_16_color_FFFFFF_t_70_pic_center 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMDA4NTM1_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/3f432f47bf724071b9b08d9b83c20d34.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwMDA4NTM1_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/d7ecec94c729490296a40457ee22b49e.png