三十分钟手撕一个线程池(保准记住线程池七大参数)

男娘i 2022-10-28 14:23 93阅读 0赞

## 1. 线程池的概念： ##

池化是我们在实际生产中经常用到的一种思想，通过一个 “池” 把资源统一的管理起来。可以达到对资源的合理管理、重复利用、减少资源创建/销毁的开销等目的。

常见的比如常量池、连接池、线程池。

线程池无非是维护一堆线程阻塞等待任务的到来，并由主线程对任务线程的数量进行动态控制的组件。做到线程资源的复用及统一管理，同时避免大量的线程创建销毁的开销，并控制总的线程数量保证系统安全。

## 1.1 新建一个线程? ##

最常用的新建线程的方式 : 得到一个新线程的方式

new Thread(new Runnable() { 
           public void run() { 
                // do sth
                // OOM : Out of Memory
           }
    }).start();

优点: 简单

缺点: 无法有效管理.

线程池的作用就是作为"线程"的调用中心, 减少其他开销.

线程池就是先创建一些线程，它们的集合称为线程池。使用线程池可以很好地提高性能.

# 2. 手撕一个简简单单的线程池 #

手写一个线程池? 对!

线程池的定义: 线程的集合.

一个线程算不算一个集合 ?

**算!**

那么有穷多个线程算不算一个集合?

**算!**

/** * 伪装线程池 * Executor :java提供的, 实现线程池的接口 * 调用实例 : * Executor executor = anExecutor; * executor.execute(new RunnableTask1()); * executor.execute(new RunnableTask2()); * * 这个接口没有强制你的实现方式,所以我们可以钻漏洞,不造线程池. ( : 反正功能我实现了.jpg */
    class DirectExecutor implements Executor { 
    
        @Override
        public void execute(Runnable r) { 
            // 直接调用run方案, 那么就是调用这个execute的线程执行这个任务了.
            r.run();
        }
    }
    
    
    
    
    /** * 那么我们来做一个最简单的线程池吧. */
    class ThreadPerTaskExecutor implements Executor { 
        public void execute(Runnable r) { 
            // 完成我们认知中最简单的线程池吧, 一个简单异步队列
            new Thread(r).start();
        }
    }

你很可能会觉得这种方式太过于简陋, 但是Doug Lea 大神在Executor 的注释中给出的代码就是这一段.

以下为Executor的源码截图.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3OTYxOTQ4_size_16_color_FFFFFF_t_70_pic_center]![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3OTYxOTQ4_size_16_color_FFFFFF_t_70_pic_center 1]  
课外扩展: 读完这段大神在源码中写的代码吧, 体验一下精妙之处在哪里.

class SerialExecutor implements Executor { 
        final Queue<Runnable> tasks = new ArrayDeque<Runnable>();
        final Executor executor;
        Runnable active;
    	// 这其实是一个可扩展的类, 你可以忽视这个构造器哦.
        SerialExecutor(Executor executor) { 
            this.executor = executor;
        }
    	
        public synchronized void execute(final Runnable r) { 
            tasks.offer(new Runnable() { 
                public void run() { 
                    try { 
                        r.run();
                    } finally { 
                        // 秒啊
                        scheduleNext();
                    }
                }
            });
            if (active == null) { 
                scheduleNext();
            }
        }
    
        protected synchronized void scheduleNext() { 
            if ((active = tasks.poll()) != null) { 
                executor.execute(active);
            }
        }
    }

# 3. 基本符合我们认知中的线程池 #

/** * 那么我们来做一个最简单的线程池吧. */
    class ThreadPerTaskExecutor implements Executor { 
        public void execute(Runnable r) { 
            // 完成我们认知中最简单的线程池吧, 一个简单异步过程
           Thread t =  new Thread(r);
           t.start();
        }
    }

虽然这个类已经基本满足了线程池的定义,但是这根本就没有任何优势,甚至因为调用方法栈的增加而额外增加了成本.

能不能让线程新建后不再被销毁?而是等待之后的任务?

就像招一群程序员,如果老板每次有了活才去招程序员, 会不会有延期交付的风险?

![点击查看源网页][ab2568914385f3f41ff479d02cbce696.png]

所以我们可以按照这个思路稍微改进一下.

import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.Executor;
    
    /** * 模拟银行工作 */
    public class BankThreadPool implements Executor { 
        /** * 只有一个柜员在工作 */
        private final CounterThread counter = new CounterThread();
    
        /** * 侯客厅, 想来办事情的任务们排队的地方 * @warning 注意这里一定要使用线程安全的队列. */
        private final BlockingQueue<Runnable> waitingRoom = new ArrayBlockingQueue<>(100);
    
        /** * 银行开工了 */
        public BankThreadPool() { 
            // 柜员们开始干活
            counter.start();
        }
    
        /** * 客人取号 * 线程池执行任务的方法，其实就是往BlockingQueue中添加元素 */
        @Override
        public void execute(Runnable task) { 
            try { 
                // 客人先取号,然后坐在侯客厅
                waitingRoom.put(task);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    
        /** * 线程池中的工作线程，直接从侯客厅中获取任务 * 然后执行任务 */
        private class CounterThread extends Thread { 
            @Override
            public void run() { 
                while (!interrupted()) { 
                    try { 
                        Runnable runnable = waitingRoom.take();//获取任务
                        if (runnable != null) { 
                            runnable.run();//执行任务
                        }
                    } catch (Exception e) { 
                        interrupt();
                        e.printStackTrace();
                    }
                }
            }
        }
    }

这代码我自己欣赏了好一会, 觉得我真牛逼, 不过还是有点细节上的问题

1.  一个柜员的银行是要倒闭了吗?
2.  可以把柜员数量, 侯客厅大小抽象出来吗? 作为接口向外提供吗?
3.  银行下班怎么说 ?

不过这些东西都是细枝末节了,大家可以考虑自己实现, 也可以直接阅读我自己提供的参考答案.

import java.util.HashSet;
    import java.util.Set;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.Executor;
    
    /** * 模拟银行工作 */
    public class BankThreadPool implements Executor { 
        //默认线程池中的线程的数量
        private static final int DEFAULT_COUNTER_NUM = 5;
        //默认处理任务的数量
        private static final int DEFAULT_WAITING_ROOM_SIZE = 100;
    
        // 线程数量
        private int counterNum;
    
        //任务数量
        private int waitingRoomSize;
    
        /** * 若干个柜员在工作 */
        private final Set<CounterThread> counters;
    
        /** * 侯客厅 */
        private final BlockingQueue<Runnable> waitingRoom;//阻塞有序队列存放任务
    
        public BankThreadPool() { 
            this(DEFAULT_COUNTER_NUM, DEFAULT_WAITING_ROOM_SIZE);
        }
    
        public BankThreadPool(int counterNum, int waitingRoomSize) { 
            // 异常清空处理
            if (counterNum <= 0) counterNum = DEFAULT_COUNTER_NUM;
            if (waitingRoomSize <= 0) waitingRoomSize = DEFAULT_WAITING_ROOM_SIZE;
            // init bank
            waitingRoom = new ArrayBlockingQueue<>(waitingRoomSize);
            this.counterNum = counterNum;
            this.waitingRoomSize = waitingRoomSize;
            counters = new HashSet<>();
    
            // 启动一定数量的线程数，从队列中获取任务处理
            for (int i = 0; i < counterNum; i++) { 
                CounterThread workThread = new CounterThread("counter_" + i);
                workThread.start();
                counters.add(workThread);
            }
        }
    
        /** * 线程池执行任务的方法，其实就是往BlockingQueue中添加元素 */
        @Override
        public void execute(Runnable task) { 
            try { 
                // 客户取号后, 加入到侯客厅中
                waitingRoom.put(task);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    
        /** * 银行下班 */
        public void closeBank() { 
            System.out.println("ready close thread pool...");
            if (counters == null || counters.isEmpty()) return;
            for (CounterThread counterThread : counters) { 
                // 每个柜员都开始下班
                counterThread.goHome();
            }
            counters.clear();
        }
    
        /** * 线程池中的工作线程，直接从BlockingQueue中获取任务 * 然后执行任务而已 * blockQueue为阻塞队列 */
        private class CounterThread extends Thread { 
            public CounterThread(String name) { 
                super();
                setName(name);
            }
    
            @Override
            public void run() { 
                // 下班了吗
                while (!interrupted()) { 
                    // 接着干
                    try { 
                        Runnable runnable = waitingRoom.take();//获取任务
                        if (runnable != null) { 
                            runnable.run();//执行任务
                        }
                    } catch (Exception e) { 
                        interrupt();
                        e.printStackTrace();
                    }
                }
                // 回家睡觉
            }
            // 回家回家, 下班不积极, 思想有问题.
            public void goHome() { 
                interrupt();
            }
        }
    }

上述代码带来了三个重大的优点:

1.  控制了线程(柜员)数量。
2.  队列 (休息室) 不但起到了缓冲(休息)的作用，还将任务的提交(取号)与执行(做事情)解耦了。 因为提交可能会有波动,但是处理的速度确实较慢的.
3.  **最重要的一点是，解决了每次重复创建和销毁线程带来的系统开销。**

# 4. 再加亿点点细节. #

上一节,我们已经有了属于自己的线程池了.

1.  真的有必要在构造器里就创建所有的柜员吗? (编制是10个, 你就要刚开张就招募十个人吗? 你这成本高呀.)
2.  当大厅坐满了, 那些取不了号的人只能回家睡觉吗? (拒绝策略)
3.  每个银行的招人策略都一样吗? 可以让银行家自己选择招人策略吗?( 构造器能加一个ThreadFactory , 让调用者自己选择thread新建策略吗)

答案自然是都可以优化的.接下来我们分情况一个一个讨论.

## 1. 真的有必要在开张时招募所有柜员吗? ##

> 线程需要占用资源的啊, 我刚上线一个APP, 没几个用户, 你就要让服务器的CPU满载吗??? 你老板不开了你.

什么时候需要线程呢? 这里我们借鉴一下现实中的企业, 当老板需要更多员工时, 他才会去招募员工.

所以我得出一个结论, 当我们的任务数较多时,再去新建一个线程.

这里说一个java底层的标准, 就是当任务数大于当前线程数, 先前线程数目周转不过来时,新建线程.

反正迟早要招够10个, 所以尽可能的让用户的体验和资源使用率达成一个均衡.(这里与ArrayList和HashMap的插入第一个元素时才申请大块数组的思想一致), 懒加载思想

// ArrayList
    // 申请数组的操作被放在第一次add时,而不是构造器中.
    void add(int x) { 
        // 如果数组没被申请,那么申请
        if (arrayNotInit()){ 
            // 申请数组内存
            initArray();
        }
    }
    //.. 同样的道理, 只有当你需要更多的线程时,才去使用

## 2. 拒绝策略 ##

> 当大厅坐满了, 那些取不了号的人只能回家睡觉吗?

答案自然是让银行家们去自定义, 每个银行的行情都不一样, 有的走高冷御姐范, 有的走大波浪热情风.

而我们只需要把这个选择权交给银行家们自己选即可.

public ThreadPool(RejectedExecutionHandler handler){ 
        this.handler = handler;
    }

// 在拒绝时调用这个策略

## 3. 每个银行的招人策略都一样吗? ##

和第二个问题相同,没有必要强制他们. 我们只需要将ThreadFactory作为一个可选的构造参数即可.

如果那家银行有自定义的需要, 传一个ThreadFactory.

public ThreadPool(RejectedExecutionHandler handler){ 
        this.handler = handler;
    }
    public ThreadPool(ThreadFactory factory){ 
        this.factory = factory;
    }
    // 组合参数
    public ThreadPool(RejectedExecutionHandler handler, ThreadFactory factory){ 
        this.handler = handler;
        this.factory = factory;
    }

根据以上三点, 把我们的代码再优化一下

import java.util.HashSet;
    import java.util.Set;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.Executor;
    import java.util.concurrent.ThreadFactory;
    
    /** * 模拟银行工作(更细节化, 更多知识点) */
    public class BankThreadPoolProPlus implements Executor { 
        /** * 默认线程池中的线程的数量 */
        private static final int DEFAULT_COUNTER_NUM = 5;
        /** * 默认处理任务的数量 */
        private static final int DEFAULT_WAITING_ROOM_SIZE = 100;
    
        /** * 线程数量 */
        private int counterMaxNum;
    
        /** * 若干个柜员在工作 */
        private final Set<Thread> counters;
    
        /** * 侯客厅 */
        private final BlockingQueue<Runnable> waitingRoom;//阻塞有序队列存放任务
        /** * 拒绝策略 */
        private volatile RejectedExecutionHandler handler;
    
        /** * 线程工厂 */
        private volatile ThreadFactory threadFactory;
    
        /** * 员工被雇佣后会得到自己的工号. */
        int nameOfCounter = 0;
    
        public BankThreadPoolProPlus() { 
            this(DEFAULT_COUNTER_NUM, DEFAULT_WAITING_ROOM_SIZE);
        }
    
        public BankThreadPoolProPlus(int counterMaxNum, int waitingRoomSize) { 
            // 异常清空处理
            if (counterMaxNum <= 0) counterMaxNum = DEFAULT_COUNTER_NUM;
            if (waitingRoomSize <= 0) waitingRoomSize = DEFAULT_WAITING_ROOM_SIZE;
            waitingRoom = new ArrayBlockingQueue<>(waitingRoomSize);
            // init bank
            this.counterMaxNum = counterMaxNum;
            counters = new HashSet<>();
        }
    
        public BankThreadPoolProPlus(int counterMaxNum, BlockingQueue<Runnable> queue,  RejectedExecutionHandler handler, ThreadFactory threadFactory) { 
                    // 异常清空处理
            if (counterMaxNum <= 0) counterMaxNum = DEFAULT_COUNTER_NUM;
    
            this.waitingRoom = queue;
            this.counterMaxNum = counterMaxNum;
            this.handler = handler;
            this.threadFactory = threadFactory;
    
        }
    
        /** * 雇佣新的雇员 */
        private void EmployeeCounter() { 
            // 如果有传入的自定义线程工厂,那么使用它的.
            Thread workThread = null;
            if (threadFactory != null) { 
                workThread = threadFactory.newThread(new Runnable() { 
                    @Override
                    public void run() { 
                        new CounterThread("counter_" + nameOfCounter++).run();
                    }
                });
            } else { 
                // 默认的新建线程的方式
                workThread = new CounterThread("counter_" + nameOfCounter++);
                workThread.start();
            }
            counters.add(workThread);
        }
    
        /** * 线程池执行任务的方法，其实就是往BlockingQueue中添加元素 */
        @Override
        public void execute(Runnable task) { 
            // 客户取号后, 加入到侯客厅中
            if (waitingRoom.offer(task)) { 
                if (waitingRoom.size() > 1 && counters.size() < counterMaxNum) { 
                    // 有超过一名顾客在等待,且编制没有用完, 那么银行加紧再雇佣一个.
                    EmployeeCounter();
                }
            } else { 
         		if (handler != null) { 
                    // 客户的任务添加失败, 说明大厅坐满人了, 那么执行拒绝策略
                    handler.rejectedExecution(task, this);
                }
            }
    
        }
    
        /** * 银行下班 */
        public void closeBank() { 
            System.out.println("ready close thread pool...");
            if (counters == null || counters.isEmpty()) return;
            for (Thread counterThread : counters) { 
                // 传入的thread应该覆写interrupt方法.
                counterThread.interrupt();
            }
            counters.clear();
        }
    
        /** * 线程池中的工作线程，直接从BlockingQueue中获取任务 * 然后执行任务而已 * blockQueue为阻塞队列 */
        private class CounterThread extends Thread { 
            public CounterThread(String name) { 
                super();
                setName(name);
            }
    
            @Override
            public void run() { 
                runBody();
                // 回家睡觉
            }
    
            private void runBody() { 
                // 下班了吗
                while (!interrupted()) { 
                    // 接着干
                    try { 
                        Runnable runnable = waitingRoom.take();//获取任务
                        if (runnable != null) { 
                            runnable.run();//执行任务
                        }
                    } catch (Exception e) { 
                        interrupt();
                        e.printStackTrace();
                    }
                }
            }
    
        }
    }

RejectHanlder的代码在这里, 简单的修改了java底层的接口

import java.util.concurrent.Executor;
    import java.util.concurrent.RejectedExecutionException;
    import java.util.concurrent.ThreadPoolExecutor;
    
    public interface RejectedExecutionHandler { 
    
        /** * Method that may be invoked by a {@link ThreadPoolExecutor} when * {@link ThreadPoolExecutor#execute execute} cannot accept a * task. This may occur when no more threads or queue slots are * available because their bounds would be exceeded, or upon * shutdown of the Executor. * * <p>In the absence of other alternatives, the method may throw * an unchecked {@link RejectedExecutionException}, which will be * propagated to the caller of {@code execute}. * * @param r the runnable task requested to be executed * @param executor the executor attempting to execute this task * @throws RejectedExecutionException if there is no remedy */
        // 改动在这里 Executor executor
        void rejectedExecution(Runnable r, Executor executor);
    }

在开启下一节之前, 我们现在已经有个好消息了, 现在的构造函数已经有四个参数了, 要知道,Java提供的ThreadPoolExercutor 也就七个接口.

这五个接口分别是

<table> 
 <thead> 
  <tr> 
   <th>字段名</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>poolsMaxSize</td> 
   <td>线程池最大线程数量</td> 
  </tr> 
  <tr> 
   <td>workQueue</td> 
   <td>任务队列</td> 
  </tr> 
  <tr> 
   <td>RejectedExecutionHandler</td> 
   <td>队列拒绝后做哪些事情,对外提供的接口</td> 
  </tr> 
  <tr> 
   <td>ThreadFactory</td> 
   <td>新建线程的工程, 对外暴漏的接口</td> 
  </tr> 
 </tbody> 
</table>

# 4. 重点来了!!! #

1.  你经历过春运和双十一吗?(当用户量集中访问时, 我们能做什么?)

春运, 运输公司采取的措施是: 加人, 加工时.

双十一, 阿里采取的措施就是加硬件, 租设备(阿里云).

而对于线程已然是满载工时, 那么要怎么做呢?

就是就使用著名的背锅侠------“临时工”.

说的专业一点, 就是"弹性思维", 那么原来有编制的柜员们, 就是核心人员了吧. 他们原来的编制数就是核心线程数.

此外, 临时工需要一个解聘的时间, 我们让银行家选择一个空闲多久可以解雇这些"临时工"…

也就是我们再原来四个参数的基础上还要加一点参数

<table> 
 <thead> 
  <tr> 
   <th>字段名</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>maxPoolSize</td> 
   <td>整个线程池最大容纳的线程个数, 也就是临时工数量= maxSize - coreSize<br>如果线程池已满,那么不会再招聘临时工了.</td> 
  </tr> 
  <tr> 
   <td>keepLiveTime</td> 
   <td>临时工(非核心线程)空闲多久就会被解聘</td> 
  </tr> 
  <tr> 
   <td>TimeUnit</td> 
   <td>keepLiveTime 的时间单位</td> 
  </tr> 
  <tr> 
   <td>poolsMaxSize -&gt; CorePoolSize</td> 
   <td>将原来的最大容量变量改为现在核心线程数量(编制数量)</td> 
  </tr> 
 </tbody> 
</table>

// 此处省略200行代码

import java.util.HashSet;
    import java.util.Set;
    import java.util.concurrent.*;
    
    /** * 模拟银行工作 * 增加临时工编制 */
    public class BankThreadPoolProPlusMaxCore implements Executor { 
        /** * 默认线程池中的线程的数量 */
        private static final int DEFAULT_CORE_NUM = 5;
        private static final int DEFAULT_COUNTER_MAX = 20;
        /** * 默认处理任务的数量 */
        private static final int DEFAULT_WAITING_ROOM_SIZE = 100;
    
        /** * 最大线程数 */
        private int maximumPoolSize;
    
        /** * 若干个柜员在工作 */
        private Set<Thread> counters = null;
    
        /** * 阻塞有序队列存放任务 * 侯客厅 */
        private final BlockingQueue<Runnable> waitingRoom;
    
        /** * 核心线程数量 */
        private volatile int corePoolSize;
    
        /** * 空闲时间 */
        private volatile long keepAliveTime;
        /** * 空闲时间单位 */
        private TimeUnit timeUnit;
        /** * 拒绝策略 */
        private volatile RejectedExecutionHandler handler;
    
        /** * 线程工厂 */
        private volatile ThreadFactory threadFactory;
    
        /** * 员工被雇佣后会得到自己的工号. */
        int nameOfCounter = 0;
    
        public BankThreadPoolProPlusMaxCore() { 
            this(DEFAULT_CORE_NUM, DEFAULT_WAITING_ROOM_SIZE);
    
        }
    
        public BankThreadPoolProPlusMaxCore(int maximumPoolSize, int waitingRoomSize) { 
            // 异常清空处理
            if (maximumPoolSize <= 0) maximumPoolSize = DEFAULT_CORE_NUM;
            if (waitingRoomSize <= 0) waitingRoomSize = DEFAULT_WAITING_ROOM_SIZE;
            waitingRoom = new ArrayBlockingQueue<>(waitingRoomSize);
            // init bank
            this.maximumPoolSize = maximumPoolSize;
            counters = new HashSet<>();
        }
    
        public BankThreadPoolProPlusMaxCore(BlockingQueue<Runnable> queue
                , RejectedExecutionHandler handler, ThreadFactory threadFactory
                , int corePoolSize, int maximumPoolSize
                , long keepAliveTime, TimeUnit timeUnit) { 
            // 异常清空处理
            if (corePoolSize <= 0) corePoolSize = DEFAULT_CORE_NUM;
            if (maximumPoolSize <= 0) maximumPoolSize = DEFAULT_COUNTER_MAX;
            this.corePoolSize = corePoolSize;
            this.keepAliveTime = keepAliveTime;
            this.maximumPoolSize = maximumPoolSize;
            this.timeUnit = timeUnit;
            this.waitingRoom = queue;
            this.handler = handler;
            this.threadFactory = threadFactory;
        }
    
        /** * 雇佣新的雇员 */
        private void employeeCounter() { 
            // 如果有传入的自定义线程工厂,那么使用它的.
            Thread workThread;
            if (threadFactory != null) { 
                workThread = threadFactory.newThread(new Runnable() { 
                    @Override
                    public void run() { 
                        new CounterThread("counter_" + nameOfCounter++).run();
                    }
                });
            } else { 
                // 默认的新建线程的方式
                workThread = new CounterThread("counter_" + nameOfCounter++);
                workThread.start();
            }
            counters.add(workThread);
        }
    
        /** * 线程池执行任务的方法，其实就是往BlockingQueue中添加元素 */
        @Override
        public void execute(Runnable task) { 
            // 客户取号后, 加入到侯客厅中
            if (waitingRoom.offer(task)) { 
                if (waitingRoom.size() > 1 && counters.size() < maximumPoolSize) { 
                    // 有超过一名顾客在等待,且编制没有用完, 那么银行加紧再雇佣一个.
                    employeeCounter();
                }
                // 否则直接放入
            } else { 
                // 侯客大厅没有位置了, 尝试紧急招募临时工工作
                if (counters.size() < maximumPoolSize) { 
                    // 当队列已满的时候(加入队列失败), 如果当前线程数小于
                    employeeCounter();
                } else if (handler != null) { 
                    // 客户的任务添加失败, 说明大厅坐满人了, 那么执行拒绝策略
                    handler.rejectedExecution(task, this);
                }
            }
    
        }
    
        /** * 银行下班 */
        public void closeBank() { 
            System.out.println("ready close thread pool...");
            if (counters == null || counters.isEmpty()) return;
            for (Thread counterThread : counters) { 
                // 传入的thread应该覆写interrupt方法.
                counterThread.interrupt();
            }
            counters.clear();
        }
    
        /** * 线程池中的工作线程，直接从BlockingQueue中获取任务 * 然后执行任务而已 * blockQueue为阻塞队列 */
        private class CounterThread extends Thread { 
            public CounterThread(String name) { 
                super();
                setName(name);
            }
    
            @Override
            public void run() { 
                runBody();
                // 回家睡觉
            }
    
            private void runBody() { 
                // 下班了吗
                while (!interrupted()) { 
                    // 接着干
                    try { 
                        // 是否是临时工(如果编制超了, 那就是临时工)
                        boolean isTempCounter = counters.size() > corePoolSize;
                        Runnable runnable;
                        if (isTempCounter) { 
                            // 尝试获取任务, 一旦超时, 那么解蛊掉
                            runnable = waitingRoom.poll(keepAliveTime, timeUnit);
                        } else { 
                            // 此处为阻塞的获取任务, 所以不会空转.
                            runnable = waitingRoom.take();//获取任务
                        }
                        if (runnable != null) { 
                            runnable.run();//执行任务
                        } else { 
                            // runnable为空, 说明临时工长期摸鱼,开除!
                            break;
                        }
                    } catch (Exception e) { 
                        interrupt();
                        e.printStackTrace();
                    }
                }
            }
        }
    }

所以最后的参数列表如下

<table> 
 <thead> 
  <tr> 
   <th>More Actions字段名</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>maxPoolSize</td> 
   <td>整个线程池最大容纳的线程个数, 也就是临时工数量= maxSize - coreSize</td> 
  </tr> 
  <tr> 
   <td>keepLiveTime</td> 
   <td>临时工(非核心线程)空闲多久就会被解聘</td> 
  </tr> 
  <tr> 
   <td>TimeUnit</td> 
   <td>keepLiveTime 的时间单位</td> 
  </tr> 
  <tr> 
   <td>corePoolSize</td> 
   <td>核心线程最大数量(编制数)</td> 
  </tr> 
  <tr> 
   <td>workQueue</td> 
   <td>任务队列:(侯客大厅)</td> 
  </tr> 
  <tr> 
   <td>RejectedExecutionHandler</td> 
   <td>拒绝的handler,对外提供的接口</td> 
  </tr> 
  <tr> 
   <td>ThreadFactory</td> 
   <td>新建线程的工程, 对外暴漏的接口(更好的支持用户自定义)</td> 
  </tr> 
 </tbody> 
</table>

注: 本文思路来自公众号"低并发编程"写的<<图解 | 你管这玩意叫线程池>>, 原文动图十分漂亮. 我只是站在他的肩膀上写了篇自己的笔记.

PS2 : 我在最后写的几个代码里, 使用了并发不安全的集合, 所以最后几个线程池有点小问题哦, 有兴趣的可以在评论区指出来.

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3OTYxOTQ4_size_16_color_FFFFFF_t_70_pic_center]: https://img-blog.csdnimg.cn/20210206170504151.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3OTYxOTQ4,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3OTYxOTQ4_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221024/893dda8aef60404aac524a18f1ea1986.png
[ab2568914385f3f41ff479d02cbce696.png]: /images/20221024/454d5d8551a541d18798461629b7a690.png