每天学习一点点20211212/20211213/20211214--线程池

线程池

- 1.线程池
- 2.JDK对线程池的支持
- 3.核心线程池的底层实现
- 4.拒绝策略
- 5.ThreadFactory
- 6.监控线程池
- 7.拓展线程池
- 8.优化线程池大小
- 9.线程池死锁
- 10.线程池中的异常处理
- 11. ForkJoinPool 线程池

1.线程池

可以以 new Thread(()->{线程执行的任务}).start(); 这种形式开启一个线程，当 run() 方法运行结束，线程对象会被 GC 释放。
在真实的生产环境中，可能需要很多线程来支撑整个应用，当线程数量非常多时，反而会耗尽 CPU 资源，如果不对线程进行控制和管理，反而会影响程序的性能。线程开销主要包括：创建与启动线程的开销；线程销毁开销；线程调度的开销；线程数量受限 CPU 处理器数量。
线程池就是有效使用线程的一种常用方式。线程池内部可以预先创建一定数量的工作线程，客户端代码直接将任务作为一个对象提交给线程池，线程池将这些任务缓存在工作队列中，线程池中的工作线程不断地从队列中取出任务并执行。

2.JDK对线程池的支持

JDK 提供了一套 Executor 框架，可以帮助开发人员有效的使用线程池。

package com.jason.java.duoxiancheng.threadpool;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


public class Test01 {
    public static void main(String[] args) {
        //创建有 5 个线程大小的线程池
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);

        //向线程池中提交 18 个任务，这 18 个任务存储到线程池的阻塞队列中，线程池中这 5 个线程就从阻塞队列中取任务执行
        for (int i = 0; i < 18; i++) {
            fixedThreadPool.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getId() + " 编号的任务在执行任务，开始时间：" + System.currentTimeMillis());
                    try {
                        Thread.sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }
}

package com.jason.java.duoxiancheng.threadpool;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;


public class Test02 {
    public static void main(String[] args) {
        //创建一个有调度功能的线程池
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);
        //在延迟2秒后执行任务，  schedule(Runnable任务，延迟时长，时间单位)
        System.out.println(System.currentTimeMillis());
        

        //以固定的频率执行任务，开启任务的时间是固定的，在 3 秒后执行任务，在上次任务结束后，在固定延迟后再次执行该任务，不管执行任务耗时多长，总是在任务结束后的 2 秒再次开启新的任务
        scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getId()+"----在固定频率开启任务---" + System.currentTimeMillis());
                try {
                    //睡眠模拟任务执行时间 ,如果任务执行时长超过了时间间隔,则任务完成后立即开启下个任务
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, 3, 2, TimeUnit.SECONDS);
    }
}

3.核心线程池的底层实现

查看 Executors 工具类中 newCachedThreadPool()，newSingleThreadExecutor()，newFixedThreadPool() 源码。

 public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue());
    }
    该线程池在极端情况下，每次提交新的任务都会创建新的线程执行，适合用来执行大量耗时短并且提交频繁的任务。

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

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new linkedBlockingQueue()));
    }

Excutors 工具类中返回线程池的方法底层都使用了 ThreadPoolExecutor 线程池，这些方法都是 ThreadPoolExecutor 线程池的封装。

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue 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;
    }

corePoolSize，指定线程池中核心线程的数量
maximumPoolSize，指定线程池中最大线程数量
keepAliveTime，当前线程池的数量超过 corePoolSize 时，多余的空闲线程的存货时长，即空闲线程在多长时长内销毁
unit，是 keepAliveTime 时长单位
workQueue，任务队列，把任务提交到该任务队列中等待执行
threadFactory，线程工厂，用于创建线程
handler，拒绝策略，当任务太多来不及处理时，如何拒绝

说明：
workQueue 工作队列是指提交未执行的任务队列，它是 BlockingQueue 接口的对象，仅用于存储 Runnable 任务。根据队列功能分类，在 ThreadPoolExecutor 构造方法中可以使用以下几种阻塞队列：
1) 直接提交队列，由 SynchronousQueue 对象提供，该队列没有容量，提交给线程池的任务不会被真实的保存，总是将新的任务提交给线程执行，如果没有空闲线程，则尝试创建新的线程，如果线程数量已经达到 maximumPoolSize 规定的最大值则执行拒绝策略。
2) 有界任务队列，由 ArrayBlockingQueue 实现，在创建 ArrayBlockingQueue 对象时，可以指定一个容量，当有任务需要执行时，如果线程池中线程数小于 corePoolSize 核心线程数则创建新的线程，如果大于 corePoolSize 核心线程数则加入等待队列，如果队列已满则无法加入，在线程数小于 maximumPoolSize 指定的最大线程数前提下会创建新的线程来执行，如果线程数大于 maxinumPoolSize 最大线程数则执行拒绝策略。
3) 无界任务队列，由 linkedBlockingQueue 对象实现，与有界队列相比，除非系统资源耗尽，否则无界队列不存在任务入队失败的情况。当有新的任务时，在系统线程数小于 corePoolSize 核心线程数则创建新的线程来执行任务，当线程池中线程数量大于 corePoolSize 核心线程数则把任务加入阻塞队列。
4) 优先任务队列是通过 PriorityBlockingQueue 实现的，是代有任务优先级的队列，是一个特殊的无界队列，不管是 ArrayBlocking 队列还是 linkedBlockingQueue 队列都是按照先进先出算法处理任务的，在 PriorityBlockingQueue 队列中可以根据任务优先级顺序先后执行。

4.拒绝策略

ThreadPoolExecutor 构造方法的最后一个参数制定了拒绝策略，当提交给线程池的任务量超过实际承载能力时，如何处理。即线程池中的线程已经用完了，等待队列也满了，无法为新提交的任务服务，可以通过拒绝策略来处理这个问题。JDK 提供了四种拒绝策略：

AbortPolicy 策略，会抛出异常
CallerRunsPolicy 策略，只要线程池没关闭，会在调用者线程中运行当前被丢弃的任务
DiscardOldestPolicy，将任务队列中最老的任务丢弃，尝试再次提交新任务
DiscardPolicy，直接丢弃这个无法处理的任务

Executors 工具类提供的静态方法返回的线程池默认的拒绝策略是 AbortPolicy 抛出异常，如果内置的拒绝策略无法满足实际需求，可以扩展 RejectedExecutionHandler 接口。

package com.jason.java.duoxiancheng.threadpool;

import java.util.Random;
import java.util.concurrent.*;


public class Test03 {
    public static void main(String[] args) {
        //定义任务
        Runnable r = new Runnable() {
            @Override
            public void run() {
                int num = new Random().nextInt(5);
                System.out.println(Thread.currentThread().getId() + "--" + System.currentTimeMillis() + "开始睡眠" + num + "秒");
                try {
                    TimeUnit.SECONDS.sleep(num);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };

        //创建线程池，自定义拒绝策略
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 5, 0, TimeUnit.SECONDS, new linkedBlockingDeque<>(10), Executors.defaultThreadFactory(), new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                //r 就是请求的任务，executor 就是当前线程池
                System.out.println(r + "id discarding...");
            }
        });
        //向线程池提交若干任务
        for (int i = 0; i < 100; i++) {
            threadPoolExecutor.submit(r);
        }
    }
}

5.ThreadFactory

线程池中的线程从哪儿来的？答案就是 ThreadFactory。
ThreadFactory 是一个接口，只有一个用来创建线程的方法： Thread newThread(Runnable r)；
当线程池中需要创建线程时就会调用该方法。

package com.jason.java.duoxiancheng.threadpool;

import java.util.Random;
import java.util.concurrent.*;


public class Test04 {
    public static void main(String[] args) throws InterruptedException {
        //定义任务
        Runnable r = new Runnable() {
            @Override
            public void run() {
                int num = new Random().nextInt(10);
                System.out.println(Thread.currentThread().getId() + "--" + System.currentTimeMillis() + "开始睡眠：" + num + "秒");
                try {
                    TimeUnit.SECONDS.sleep(num);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        //创建线程池，使用自定义线程工厂，采用默认的拒绝策略是抛出异常
        ExecutorService executorService = new ThreadPoolExecutor(5, 5, 0, TimeUnit.SECONDS, new SynchronousQueue<>(), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                //根据参数 r 接收的任务，创建一个线程
                Thread t = new Thread(r);
                t.setDaemon(true);//设置为守护线程，当主线程运行结束，线程池中的线程会自动退出
                System.out.println("创建了线程：" + t);
                return t;
            }
        });

        //提交5个任务，当给当前线程池提交的任务超过5个是，线程池默认抛出异常
        for (int i = 0; i < 5; i++) {
            executorService.submit(r);
        }

        //主线程睡眠
        Thread.sleep(10000);
        //主线程睡眠超时，主线程结束，线程池中的线程会自动退出
    }
}

6.监控线程池

ThreadPoolExecutor 提供了一组方法用于监控线程池

7.拓展线程池 8.优化线程池大小 9.线程池死锁 10.线程池中的异常处理 11. ForkJoinPool 线程池

每天学习一点点20211212/20211213/20211214--线程池

Java相关栏目本月热门文章