从ThreadPoolExecutor看线程池

矫情吗；* 2022-12-20 06:12 141阅读 0赞

**首先复习下创建线程的几种方式**

1、实现runnable接口
    
    	new Thread(() -> log.info("方式一：实现runnable接口")).start();
    	
    2、实现callable接口
    
    	FutureTask<String> task = new FutureTask<>(() -> "方式二：实现callable接口");
        new Thread(task).start();
    
    3、继承Thread类
    
    	class MyThread extends Thread{ 
    	    @Override
    	    public void run() { 
    	        log.info("方式三：继承thread类");
    	    }
    	};
    
    4、使用线程池  这个只是创建了线程池，但是线程具体需要执行的任务还是需要在方法中去指定
    
    	ExecutorService executorService = Executors.newCachedThreadPool();
        executorService.submit(() -> log.info("方式四：使用线程池"));

![在这里插入图片描述][202011111454285.png_pic_center]  
**ok 现在我们开始来熟悉线程池**

**1、创建线程池**

public ThreadPoolExecutor(int corePoolSize, //核心线程数
                              int maximumPoolSize, //最大线程数
                              long keepAliveTime, //存活时间
                              TimeUnit unit, //时间单位
                              BlockingQueue<Runnable> workQueue, //阻塞队列，用来储存线程
                              ThreadFactory threadFactory, //创建线程的工厂
                              RejectedExecutionHandler handler //拒绝策略 ) { 
    }
    
    //具体代码引发的问题
    
    public class ThreadPoolListener implements BeanPostProcessor , InitializingBean { 
        //设置阻塞队列大小为2
        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(2);
        //设置核心线程数为4 最大线程数为6
        ThreadPoolExecutor pool = new ThreadPoolExecutor(4, 6, 10, TimeUnit.SECONDS, queue, (runnable, executor) -> { 
            //拒绝策略需要处理的业务逻辑
            log.info("do something ~");
            log.info(runnable.toString());
            log.info(executor.toString());
        });
        //定时任务处理业务~~~
        @Scheduled(cron = "*/5 * * * * *")
        @Async
        public void logCollection(){ 
            //打印线程池的工作中的线程数以及线程id 队列的大小
            pool.execute(() ->{ 
                log.info("测试线程池工作原理 线程数:{},当前线程id:{}",pool.getPoolSize(),Thread.currentThread().getId());
                log.info(queue.size()+"----->>队列大小");
                if(queue.size()>0){ 
                    log.info("队列中的数据:{}",queue.peek());
                }
                try { 
                    Thread.sleep(100000);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            });
        }
        @Override
        public void afterPropertiesSet() throws Exception { 
            log.info("属性设置完毕 队列容量:{}",queue.size());
        }
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMDAxOTQ1_size_16_color_FFFFFF_t_70_pic_center]  
从图中可以看到，在线程数量超过了核心线程之后，队列立马就满了 这是为什么?

从execute入手，看看他都做了什么事情

public void execute(Runnable command) { 
        if (command == null)
            throw new NullPointerException();
        //ctl:原子计数器，来保存当前的正在执行的线程数以及线程状态（高3位存状态，低29位存线程数）
        int c = ctl.get();
        //执行中的线程数是否小于核心线程数
        if (workerCountOf(c) < corePoolSize) { 
            //添加一个worker来处理任务
            if (addWorker(command, true))
                return;
            //走到这里说明worker添加失败了 再次获取下线程的状态数
            c = ctl.get();
        }
        //如果线程池是运行的状态，将本任务加入到队列中
        if (isRunning(c) && workQueue.offer(command)) { 
            int recheck = ctl.get();
            //再判断线程池是不是运行中 不是的话，将本线程移出队列
            if (! isRunning(recheck) && remove(command))
                // 然后执行拒绝策略
                reject(command);
            //线程数量为0的话， 开启创建一个新的worker 注意传的参数
            else if (workerCountOf(recheck) == 0)
                //开启一个新worker，去处理队列中的任务 这里false，在方法里面会作为判断允许线程的数量的条件
                //上面一行话看不懂，可以去看看这个方法，就自然懂了
                addWorker(null, false);
        }
        //线程池不是运行中 或者任务加入队列失败 再次尝试添加一个worker去处理任务
        else if (!addWorker(command, false))
            //添加失败，执行拒绝策略
            reject(command);
    }

从上面我们可以看到，在判断完线程池的状态以及数量之后，要么执行拒绝策略，要么创建一个worker，要么将任务加入到队列中，所以具体是怎么去处理线程任务？我们来分析addWorker

private boolean addWorker(Runnable firstTask, boolean core) { 
      retry:
        for (;;) { 
            int c = ctl.get();
            int rs = runStateOf(c);
    
           //如果线程已经关闭 并且 任务为null，队列不为空 直接返回 
           //（因为线程已经关闭，任务队列也为空，当前也没有需要执行的任务。所以直接返回）
           //(如果任务队列不为空，或者当前需要执行的任务不为null，并且线程池状态为shutdown，
           //线程池会执行完当前的任务，如果状态为stop，则直接返回，所有任务都会被抛弃)
            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();  
                //再次判断下线程池状态 如果状态被改变了，那么执行下一次的外层循环 在去获取线程池状态、数量等信息
                if (runStateOf(c) != rs)
                    continue retry;
            }
        }
    
        //到了这里，说明已经成功增加了线程数量，才会跳出上面的循环
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try { 
        	//增加一个新的worker，将我们传过来的任务传进去，作为线程的第一个任务去执行
        	//worker中会创建一个线程，一个worker对应一个线程
            w = new Worker(firstTask);
            //拿到worker中的线程
            final Thread t = w.thread;
            if (t != null) { 
                final ReentrantLock mainLock = this.mainLock;
                //加独占锁 这一块就与aqs关联起来了 怎么加锁，查看我前面的文章
                mainLock.lock();
                try { 
                    int rs = runStateOf(ctl.get());
                    //线程池是运行中的状态 或者是shutdown，并且本任务不为null
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) { 
                        //本线程已经在运转，直接抛异常
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                        //将这个worker加入到工作组中
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally { 
                    //释放锁 看到这里，我们可以知道，
                    //上面枷锁，只是为了保证在添加woker到工作组中，线程池的状态不被改变
                    mainLock.unlock();
                }
                //worker被成功加入到工作组中了。再来启动这个worker去工作
                if (workerAdded) { 
                    //重点！！！
                    t.start();
                    workerStarted = true;
                }
            }
        } finally { 
            if (! workerStarted)
                //worker没启动，需要做一些清理工作，如前面 workCount 加了 1，将其减掉
                addWorkerFailed(w);
        }
        //返回worker的工作状态
        return workerStarted;
    }

我们都知道，在调用了thread.start()方法之后，他会启动线程，然后去执行线程的run方法。同样，我们在将worker启动之后，也会调用他的run方法。我们来看run里面都做了什么

这里做一个简单说明：在创建worker的时候，他们将我们需要执行的任务作为必要的构造器参数传进去，在构造方法里面，会对他做一个包装之后，在赋给worker中的thread字段。

public void run() { 
        runWorker(this);
    }
    
    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 或者为null，那么就去取任务
            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(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 { 
            //到这里有两种情况
            //1 任务为空 那么正常走到这里completedAbruptly 已被改为false
            //2 出现了异常 此时completedAbruptly由于异常，所以他的值还是true
            processWorkerExit(w, completedAbruptly);
        }
    }

任务为空 或者出现异常之后，来在这里做处理

private void processWorkerExit(Worker w, boolean completedAbruptly) { 
        //如果有异常，那么减少线程的数量
        if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
            decrementWorkerCount();
    
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try { 
            completedTaskCount += w.completedTasks;
            //将本worker移除工作组
            workers.remove(w);
        } finally { 
            mainLock.unlock();
        }
        //更改线程池状态 略过吧 不是重点 感兴趣的可以自己看 也很简单 反正我没看
        tryTerminate();
    
        int c = ctl.get();
        //如果线程池的状态<stop
        if (runStateLessThan(c, STOP)) { 
            //没出现异常
            if (!completedAbruptly) { 
                int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
                //队列不为空，则保证最少有一个线程存在，来处理这个队列
                if (min == 0 && ! workQueue.isEmpty())
                    min = 1;
                if (workerCountOf(c) >= min)
                    return; // replacement not needed
            }
            addWorker(null, false);
        }
    }

再来看看 线程是如何获取任务的

private Runnable getTask() { 
        boolean timedOut = false; // Did the last poll() time out?
    
        for (;;) { 
            int c = ctl.get();
            int rs = runStateOf(c);
    		//线程池状态>=stop 或者线程池状态=shutdown，队列为空 也是再次保证，线程shutdown之后，不在接收新的创建线程然后去处理任务的请求 但是已经存在的任务还是可以正常去处理
            if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { 
                //减少线程的数量
                decrementWorkerCount();
                return null;
            }
    
            int wc = workerCountOf(c);
    		//正在工作的线程是否超过核心线程数
            boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
            //队列为空 或者线程池状态改变 那么也返回null
            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) { 
            	// 如果此 worker 发生了中断，采取的方案是重试
                // 解释下为什么会发生中断，这个读者要去看 setMaximumPoolSize 方法。
    
                // 如果开发者将 maximumPoolSize 调小了，导致其小于当前的 workers 数量，
                // 那么意味着超出的部分线程要被关闭。重新进入 for 循环，自然会有部分线程会返回 null
                timedOut = false;
            }
        }
    }

至此，关于任务是怎么加入到队列，以及又是怎么被取出来然后去执行的已经很清晰了

其他一些细节点  
**corePoolSize**

核心线程数，不要抠字眼，反正先记着有这么个属性就可以了。

**maximumPoolSize**

最大线程数，线程池允许创建的最大线程数。

**workQueue**

任务队列，BlockingQueue 接口的某个实现（常使用 ArrayBlockingQueue 和 LinkedBlockingQueue）。

**keepAliveTime**

空闲线程的保活时间，如果某线程的空闲时间超过这个值都没有任务给它做，那么可以被关闭了。注意这个值并不会对所有线程起作用，如果线程池中的线程数少于等于核心线程数 corePoolSize，那么这些线程不会因为空闲太长时间而被关闭，当然，也可以通过调用 allowCoreThreadTimeOut(true)使核心线程数内的线程也可以被回收。

**threadFactory**

用于生成线程，一般我们可以用默认的就可以了。通常，我们可以通过它将我们的线程的名字设置得比较可读一些，如 Message-Thread-1， Message-Thread-2 类似这样。

**handler：**

当线程池已经满了，但是又有新的任务提交的时候，该采取什么策略由这个来指定。有几种方式可供选择，像抛出异常、直接拒绝然后返回等，也可以自己实现相应的接口实现自己的逻辑，这个之后再说。

参考文章：https://javadoop.com/post/java-thread-pool\#toc\_4

[202011111454285.png_pic_center]: /images/20221120/936bfc27b0a64397b18ef4f92c30a13c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMDAxOTQ1_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/20201111165334436.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxMDAxOTQ1,size_16,color_FFFFFF,t_70#pic_center