线程池

2019-11-01 本文已影响0人木木不哭_8be4

1线程池的使用及其优势

1.1 为什么要使用线程池

降低资源的消耗：通过复用已经创建好的线程来降低重复创建线程和消耗线程所带来的资源消耗
提高响应速度：任务来临的时候，不需要等待线程的创建就可以直接使用
提高线程的可管理性：线程是稀缺资源，无限制的创建线程会造成不仅会不必要的系统资源损耗，还会降低系统的稳定性。使用线程池可以很直观的控制最大线程数，对线程进行统一的监控，也方便了调优工作。

1.2 jdk提供的常见线程池

       //定容的线程池
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);
        //仅允许一个线程同时存在的线程池
        ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
        //不固定容量的线程（会随着任务数的增加而改变活跃线程数）
        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();

1.3 常见线程池的使用

分别定义了三个不同类型的线程池，使用for循环模拟10个任务请求。
观察打印的结果可以发现：
Executors.newFixedThreadPool(5)最多有5个线程工作
Executors.newSingleThreadExecutor()最多有1个线程工作
Executors.newCachedThreadPool()根据任务量的多少来动态产生线程数

public class NormolThreadpool {
    public static void main(String[] args) {
        String yeu = "办理业务";
        //定容的线程池
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);
        //仅允许一个线程同时存在的线程池
        ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
        //不固定容量的线程（会随着任务数的增加而改变活跃线程数）
        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
        System.out.println("----------FixedThreadPool----------------");
        try {
            for (int i = 1; i <= 10; i++) {
                fixedThreadPool.submit(() -> {
                    System.out.println(Thread.currentThread().getName() + yeu);
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //但凡涉及到池之类的资源 都要记得关闭
            fixedThreadPool.shutdown();
        }

        try {
            TimeUnit.SECONDS.sleep(1L);

        } catch (Exception e) {

            e.printStackTrace();
        }

        System.out.println("----------CacheThreadPool----------------");
        try {
            for (int i = 1; i <= 10; i++) {
                cachedThreadPool.submit(() -> {
                    System.out.println(Thread.currentThread().getName() + yeu);
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //但凡涉及到池之类的资源 都要记得关闭
            cachedThreadPool.shutdown();
        }


        try {
            TimeUnit.SECONDS.sleep(1L);

        } catch (Exception e) {

            e.printStackTrace();
        }

        System.out.println("----------SingleThreadPool----------------");
        try {
            for (int i = 1; i <= 10; i++) {
                singleThreadExecutor.submit(() -> {
                    System.out.println(Thread.currentThread().getName() + yeu);
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //但凡涉及到池之类的资源 都要记得关闭
            singleThreadExecutor.shutdown();
        }
    }
}

打印结果：

----------FixedThreadPool----------------
pool-1-thread-1办理业务
pool-1-thread-1办理业务
pool-1-thread-1办理业务
pool-1-thread-1办理业务
pool-1-thread-1办理业务
pool-1-thread-1办理业务
pool-1-thread-2办理业务
pool-1-thread-3办理业务
pool-1-thread-4办理业务
pool-1-thread-5办理业务
----------CacheThreadPool----------------
pool-3-thread-1办理业务
pool-3-thread-2办理业务
pool-3-thread-3办理业务
pool-3-thread-4办理业务
pool-3-thread-5办理业务
pool-3-thread-1办理业务
pool-3-thread-2办理业务
pool-3-thread-1办理业务
pool-3-thread-5办理业务
pool-3-thread-4办理业务
----------SingleThreadPool----------------
pool-2-thread-1办理业务
pool-2-thread-1办理业务
pool-2-thread-1办理业务
pool-2-thread-1办理业务
pool-2-thread-1办理业务
pool-2-thread-1办理业务
pool-2-thread-1办理业务
pool-2-thread-1办理业务
pool-2-thread-1办理业务
pool-2-thread-1办理业务

1.4 常见线程池的源码分析

public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

可以发现他们底层都调用了一个ThreadPoolExecutor（线程池的底层实现都要靠这个）类的构造函数，这个构造函数的参数先不谈。
同时还可以注意到newFixedThreadPool 和newSingleThreadExecutor的第五个参数都用到了LinkedBlockingQueue<Runnable>,而jdk对它的描述是

Creates a LinkedBlockingQueue with a capacity of Integer.MAX_VALUE.`

进一步的深究：

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler)

可以发现：

真正用来做实际事情的是这个7个参数的构造方法

2.线程池的7大参数（以银行为例）

corePoolSize:池内核心线程数，相当于银行中当前值班的窗口数量
maximumPoolSize:池内最大线程数---银行中所有的窗口数量，包含corePoolSize
keepAliveTime:保持线程活跃时间。
unit:时间单位
workQueue:阻塞队列 ----办理业务时窗口都有人，在休息区等待的人
threadFactory:线程工厂
handler:等待队列也已经排满了,再也塞不下新的任务了同时,线程池的max也到达了,无法接续为新任务服务。这时我们需要拒绝策略机制合理的处理这个问题.

2.1线程池的底层原理（执行流程）

假定 corePoolSize=2 maximumPoolSize=5 keepAliveTime=2L unit=TimeUnit.SECENDS
workQueue的容量为3

最开始任务提交的时候，线程池直接接收请求。
当corePoolSize满了之后，如果还有新的任务请求，会判断workQueue是否已满，
如果workQueue没满就让任务请求进去阻塞队列等待
如果workQueue也满了，就会激活新的线程，总线程数不能超过maximumPoolSize
如果还有新的任务请求进来同时 maximumPoolSize也满了，那就需要用到拒绝策略
自定义的线程池案例：

public static void main(String[] args) {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                2,
                5,
                1L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(3),
                Executors.defaultThreadFactory(),
                //抛异常的拒绝策略RejectedExecutionException
                //new ThreadPoolExecutor.AbortPolicy()
                //把任务交还给调用此线程的线程去执行
                //new ThreadPoolExecutor.CallerRunsPolicy()
                //抛弃队列中等待最久的任务，然后把自己加入队列
                //new ThreadPoolExecutor.DiscardOldestPolicy()
                //直接丢弃任务，如果任务可用丢弃这是最好的策略
                new ThreadPoolExecutor.DiscardPolicy()
        );
        try {
            //模拟十个任务 需要开启10个线程
            for (int i = 0; i < 10; i++) {
                threadPoolExecutor.execute(() -> {
                    System.out.println(Thread.currentThread().getName() + "处理中---");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPoolExecutor.shutdown();
        }

    }

2.2 线程池的选择

都不用，参考阿里巴巴开发手册。

【强制】线程池不允许使用 Executors 去创建，而是通过 ThreadPoolExecutor 的方式，这样
的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。
说明： Executors 返回的线程池对象的弊端如下：
1） FixedThreadPool 和 SingleThreadPool:
允许的请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM。
2） CachedThreadPool 和 ScheduledThreadPool:
允许的创建线程数量为 Integer.MAX_VALUE，可能会创建大量的线程，从而导致 OOM。

2.3 如何合理的配置线程池

2.3.1 CPU密集型

说明：需要大量运算没有阻塞，cpu一直全速运行（单核cpu的话无论开几个线程都一样）
公式：CPU核数+1 个线程数尽可能少的线程数
Runtime.getRuntime().availableProcessors() 获取cpu核心数

2.3.2 IO密集型

因为需要大量的io，意味着大量的IO所以要尽可能多的线程。
（cpu数-1）/阻塞系数（0.8-0.9）
或者
cpu*2

线程池