【多线程】线程池 详解

1. 线程池是什么

虽然线程的创建和销毁的开销比较小, 但还是有的, 如果频繁的创建和销毁线程, 开销还是比较大的.解决: 线程池或者协程, 本文主讲线程池.

线程池: 把线程提前创建好, 放到池子里, 后面需要用到线程直接从池子里面取, 不必从系统申请, 
线程用完, 不是还给系统, 而是放到池子里面, 必备下次使用, 这样就减去了频繁创建销毁线程的开销.
1
2

举个栗子:

• 一家快递店，店里没有雇人，而是每次有业务来了，就现场找一名同学过来把快递送了，然后解雇同学。这个类比我们平时来一个任务，起一个线程进行处理的模式。
• 很快老板发现问题来了，每次招聘 + 解雇同学的成本还是非常高的。
• 老板还是很善于变通的，于是决定最多雇10个人, 并且雇了 10 个人之后不再解雇, 当一个任务来时, 老板就看如果还没有雇到 10 个(就算已经雇的人是闲着的, 只要没雇到 10 个就会雇新人), 就雇 1 个人去送快递, 否则只是把业务放到一个本本上，等这 10 个快递人员有空闲的时候去处理。

这个就是我们要带出的线程池的模式。
线程池最大的好处就是减少每次启动、销毁线程的损耗。

为什么线程放到池子里面就比从系统申请释放更快 ?
这就涉及到操作系统的用户态与内核态:

自己写的代码运行在用户态, 有些代码需要通过调用操作系统的 API, 进一步的逻辑会在内核中执行.
比如 System.out.println, 本质上需要经过 write 系统调用进入到内核中, 在内核里面会执行一些逻辑, 在内核中运行的代码称为 “内核态” 运行的代码.

创建线程本身需要内核的支持, 创建线程本身就是在操作系统内核中创建 PCB, 调用 Thread.start 也要进入内核态运行.

而把创建好的线程放到 “池子里”, 由于 “池子” 是用户态实现的, 放到池子里/从池子里取这个过程不需要涉及到内核态, 用纯粹的用户代码就能完成, 一般认为纯用户态的操作效率比经过内核态处理的操作效率更高.
为什么 ?
不是说内核处理的效率一定真的低, 而是说代码进入内核态之后运行就不可控了, 你不知道此时内核背负了多少任务, 内核什么时候执行我们的任务, 什么时候运行完就不知道了, 有时快有时慢, 所以是不可控的.

2. 标准库中的线程池

1. ThreadPoolExecutor
   构造方法:

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
1
2
3
4
5
6
7

• corePoolSize: 核心线程数 (正式员工数量)
• maximumPoolSize: 最大线程数 (正式员工 + 临时工数量)
• keepAliveTime: 闲置的线程数超过corePoolSize时, 临时工多长时间后被销毁
• unit: 时间的单位 (s, ms, us)
• workQueue: 任务队列, 通过 submit 方法将任务放到任务队列中
• threadFactory: 线程工厂, 线程是怎么创建出来的
• handler: 拒绝策略, 任务队列满了怎么办? 1. 忽略最新任务 2. 阻塞等待 3. 丢弃最老的任务 4. 抛异常…

2. Executors
   标准库中简化版本的线程池, 本质是针对 ThreadPoolExecutor 进行封装, 还提供一些默认参数.

• 使用 Executors.newFixedThreadPool(10) 能创建出固定包含 10 个线程的线程池.
• 返回值类型为 ExecutorService
• 通过 ExecutorService.submit 可以注册一个任务到线程池中.

        ExecutorService pool = Executors.newFixedThreadPool(10);
        pool.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("hello");
            }
        });
1
2
3
4
5
6
7

Executors 创建线程池的几种方式:

• newFixedThreadPool: 创建固定线程数的线程池
• newCachedThreadPool: 创建线程数目动态增长的线程池.
• newSingleThreadExecutor: 创建只包含单个线程的线程池.
• newScheduledThreadPool: 设定 延迟时间后执行命令，或者定期执行命令. 是进阶版的 Timer.
• newSingleThreadScheduledExecutor: 创建一个单个线程带有定时器的线程池.
• newWorkStealingPool: 创建一个任务抢占式的线程池.每个线程都有自己的任务队列, 当把自己的任务全执行完了, 就去其他线程的任务队列中 “窃取” 任务进行执行.(在日常生活中的话就是指帮助别人干一些工作)

所以加上 ThreadPoolExecutor 共有 7 种创建线程池的方法

3. 实现线程池

• 核心操作为 submit, 将任务加入线程池中
• 使用 Worker 类描述一个工作线程. 使用 Runnable 描述一个任务.
• 使用一个 BlockingQueue 组织所有的任务
• 每个 worker 线程要做的事情: 不停的从 BlockingQueue 中取任务并执行.
• 指定一下线程池中的最大线程数 maxWorkerCount; 当当前线程数超过这个最大值时, 就不再新增线程了.

class MyThreadPool {
    //任务直接用Runnable,不用另外创建类了
    //用阻塞队列来组织任务
    private BlockingQueue<Runnable> queue=new LinkedBlockingQueue<>();

    //用来描述线程的类
    static class Worker extends Thread{
        private BlockingQueue<Runnable> queue=null;
        //利用构造方法获取任务队列
        public Worker(BlockingQueue<Runnable> queue){
            this.queue=queue;
        }

        @Override
        public void run() {
            while(true){
                try {
                    //如果队列为空则阻塞
                    Runnable runnable=queue.take();
                    runnable.run();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    //用数组来当作池存储线程
    List<Thread> list=new ArrayList<>();
    //构造方法中创建若干线程放到线程池中
    public MyThreadPool(int n){
        for(int i=0;i<n;i++){
            Worker worker=new Worker(this.queue);
            //注意不要忘记start
            worker.start();
            //加入线程池中
            list.add(worker);
        }
    }

    //提交任务
    public void submit(Runnable runnable)  {
        try {
            this.queue.put(runnable);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public static void main(String[] args) {
        MyThreadPool pool=new MyThreadPool(10);
        for(int i = 1;i <= 100;i++){
            int n = i;
            //submit了100次,相当于100个任务进入了任务队列,每个线程分一些,很快就执行完了
            pool.submit(new Runnable() {
                @Override
                public void run() {
                    System.out.println("myThreadPool: " + n);
                }
            });
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

4. 面试题

有一个程序, 这个程序需要并发的完成一些任务, 如果使用线程池的话, 线程池中的线程数设置为多少合适 ?

回答:

• 我们要通过性能测试, 才能找到合适的值.
• 例如写一个服务器程序, 服务器通过线程池处理用户请求, 就可以针对这个服务器进行性能测试, 比如构造一些请求发送给服务器, 要进行性能测试的话, 请求就要构造很多, 比如 每秒发送 500/1000/2000 个, 可以根据实际业务场景构造一个合适的值
• 然后根据线程池中不同的线程数, 观察任务的处理速度, 程序持有的 CPU 占有率, 一般线程数越多, 执行速度越快, 但是 CPU 占有率越高, 需要找到一个程序速度能接收且 CPU 占有率也合理的平衡点.

为什么不让 CPU 占有率 过高 ?

对于线上服务器, 一定要留有一定的冗余以便随时应对可能的突发情况, 例如请求暴涨, 若本身就把 CPU 快占完了, 这是突然来了一波请求高峰,此时服务器可能直接就挂了.

总结: 自己实现线程池

1. 能够描述任务 (Runnable)
2. 能够组织任务 (BlockingQueue)
3. 能够描述工作线程
4. 组织线程
5. 线程从任务队列中取任务执行