• 【并发编程】线程池及Executor框架

    文章目录

    1.为什么要使用线程池
       诸如 Web 服务器、数据库服务器、文件服务器或邮件服务器之类的许多服务器应用程序都面向处理来自某些远程来源的大量短小的任务。请求以某种方式到达服务器，这种方式可能是通过网络协议（例如 HTTP、FTP ）、通过 JMS队列或者可能通过轮询数据库。不管请求如何到达，服务器应用程序中经常出现的情况是：单个任务处理的时间很短而请求的数目却是巨大的。每当一个请求到达就创建一个新线程，然后在新线程中为请求服务，但是频繁的创建线程，销毁线程所带来的系统开销其实是非常大的。

    • 1
        线程池为线程生命周期开销问题和资源不足问题提供了解决方案。通过对多个任务重用线程，线程创建的开销被分摊到了多个任务上。其好处是，因为在请求到达时线程已经存在，所以无意中也消除了线程创建所带来的延迟。这样，就可以立即为请求服务，使应用程序响应更快。而且，通过适当地调整线程池中的线程数目，也就是当请求的数目超过某个阈值时，就强制其它任何新到的请求一直等待，直到获得一个线程来处理为止，从而可以防止资源不足。

    • 1
    风险与机遇:
用线程池构建的应用程序容易遭受任何其它多线程应用程序容易遭受的所有并发风险，
诸如同步错误和死锁，它还容易遭受特定于线程池的少数其它风险，诸如与池有关的死锁、资源不足和线程泄漏。

    • 1
    • 2
    • 3
    2.线程池创建线程

    • Java通过Executors提供四种线程池

      • newCachedThreadPool创建一个可缓存的线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。
      • newFixedThreadPool创建一个定长的线程池，可以控制线程的最大并发数，超出的线程会在队列中等待。
      • newScheduledThreadPool创建一个定长的线程池，支持定时及周期性任务执行。
      • newSingleThreadExecutor创建一个单线程的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。

    • 线程代码

    public class ThreadForPools implements Runnable {

    private Integer index;

    public ThreadForPools(Integer index) {
        this.index = index;
    }

    @Override
    public void run() {
        try {
            System.out.println("开始处理线程");
            Thread.sleep(index*100);
            System.out.println("我的线程标识是"+this.toString());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    • newCachedThreadPool

    • 可以有无限多的线程进来（线程地址不一样），但是需要注意机器的性能。

    /**
 * 创建可缓存的线程池
 */
public class MyCachedThreadPool {
    public static void main(String[] args) {
        ExecutorService cachedThreadPool = Executors.newCachedThreadPool();

        for (int i = 0; i < 5; i++) {
            cachedThreadPool.execute(new ThreadForPools(i));
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

    在这里插入图片描述

    • newFixedThreadPool
    • 每次最多只有指定个线程在处理，当第一批线程执行完毕后，新的线程进来进行处理（线程地址不一样）。
    public class MyFixedThreadPool {
    public static void main(String[] args) {
        //线程池允许同时存在两个线程
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(2);
        for (int i = 0; i < 5; i++) {
            fixedThreadPool.execute(new ThreadForPools(i));
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    在这里插入图片描述

    • newScheduledThreadPool
    • 创建一个定长的线程池，支持定时周期性任务执行。
    //schedule(commod,delay,unit)这个方法是说明系统启动后，需要等待多久时间执行，delay是等待的时间，只执行一次没有周期性。

public class MyScheduledThreadPool {
    public static void main(String[] args) {
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(2);

        for (int i = 0; i < 5; i++) {
            scheduledThreadPool.schedule(new ThreadForPools(i),5,TimeUnit.SECONDS);
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

    在这里插入图片描述

    //scheduleAtFixedRate(commod,initialDelay,period,unit),这个是以period周期性执行任务，initialDelay是系统启动等待时间。

public class MyScheduledThreadPool {
    public static void main(String[] args) {
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);

        for (int i = 0; i < 5; i++) {
             //线程启动等待2秒执行，之后每3秒执行一个周期
            scheduledThreadPool.scheduleAtFixedRate(new ThreadForPools(i),2,3,TimeUnit.SECONDS);
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

    在这里插入图片描述

    //scheduleWithFixedDelay(commod,initialDelay,period,unit),这个是以period周期性执行任务，initialDelay是系统启动等待时间,和scheduleAtFixedRate的区别在于系统等待的时间不记在周期性执行的时间内。

public class MyScheduledThreadPool {
    public static void main(String[] args) {
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);

        for (int i = 0; i < 5; i++) {
             //线程启动等待2秒执行，之后每3秒执行一个周期
            scheduledThreadPool.scheduleWithFixedDelay(new ThreadForPools(i),5,3,TimeUnit.SECONDS);
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

    在这里插入图片描述

    • newSingleThreadExecutor
    • 创建一个单线程化的线程池，他只会用唯一的工作线程来执行任务，保证所有任务按照顺序（FIFO、LIFO、优先级）执行。
    public class MySingleThreadExecutor {
    public static void main(String[] args) {
        ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 5; i++) {
            singleThreadExecutor.execute(new ThreadForPools(i));
        }
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    在这里插入图片描述

    3.ThreadPoolExecutor类
    • java.util.concurrent.ThreadPoolExecutor类时线程池中最核心的一个类，因此如果要彻底了解java中的线程池，必须先了解这个类。下面我们来看一下ThreadPoolExecutor类的具体实现源码。
    public class ThreadPoolExecutor extends AbstractExecutorService {
    .....
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
            BlockingQueue<Runnable> workQueue);
 
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
            BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory);
 
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
            BlockingQueue<Runnable> workQueue,RejectedExecutionHandler handler);
 
    public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,
        BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler);
    ...
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 从上面的代码可以得知，ThreadPoolExecutor继承了AbstractExecutorService类，并提供了四个构造器，事实上，通过观察每个构造器的源码具体实现，发现前面三个构造器都是调用的第四个构造器进行的初始化工作。
    • 下面解释下各个参数的含义：
    • corePoolSize：核心池的大小，这个参数跟后面讲述的线程池的实现原理有非常大的关系。在创建了线程池后，默认情况下，线程池中并没有任何线程，而是等待有任务到来才创建线程去执行，除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法，从这两个方法的名字就可以看出，是预创建线程的意思，记载没有任务到来之前就创建corePoolSize个线程或一个线程。默认情况下，在创建了线程池后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把达到的任务方到缓存队列中。
    • maximumPoolSize：线程池中最大的线程数，这个参数也是一个非常重要的参数，它表示在线程池中最多能创建多个个线程。
    • keepAliveTime：表示线程没有任务执行时最多能保持多久时间会终止。默认情况下当线程池中的线程数大于corePoolSize时，keepAliveTime才会起作用，直到线程池中的线程数不大于corePoolSize，keepAliveTime就不会起作用，即不会对初始化预创建的线程起作用。但是如果调用了allowCoreThreadTimeOut(boolean)方法，keepAliveTime也会对预创建的线程起作用，直至线程池中线程为0。
    • unit：参数keepAliveTime的时间单位，有7种取值
      • TimeUnit.DAYS; //天
      • TimeUnit.HOURS; //小时
      • TimeUnit.MINUTES; //分钟
      • TimeUnit.SECONDS; //秒
      • TimeUnit.MILLISECONDS; //毫秒
      • TimeUnit.MICROSECONDS; //微妙
      • TimeUnit.NANOSECONDS; //纳秒
    • workQueue：一个阻塞队列，用来存储等待执行的任务，这个参数的选择也很重要，会对线程池的运行过程产生重大影响，一般来说，这里的阻塞队列有以下几种选择：
      • ArrayBlockingQueue
      • LinkedBlockingQueue
      • SynchronousQueue
    • threadFactory：线程工厂，主要用来创建线程。
    • handler：表示当拒绝处理任务时的策略，有以下四种取值：
      • ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
      • ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
      • ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
      • ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务
    • 类关系图

    在这里插入图片描述

    Executor是一个顶层接口，在它里面只声明了一个方法execute(Runnable)，返回值为void，参数为Runnable类型，从字面意思可以理解，就是用来执行传进去的任务的；然后ExecutorService接口继承了Executor接口，并声明了一些方法：submit、invokeAll、invokeAny以及shutDown等；抽象类AbstractExecutorService实现了ExecutorService接口，基本实现了ExecutorService中声明的所有方法；然后ThreadPoolExecutor继承了类AbstractExecutorService。

    • 1
    • 在ThreadPoolExecutor类中有几个非常重要的方法：
      • execute()
      • submit()
      • shutdown()
      • shutdownNow()
    • execute()：实际上是Executor中声明的方法，在ThreadPoolExecutor进行了具体的实现，这个方法是ThreadPoolExecutor的核心方法，通过这个方法可以向线程池提交一个任务，交由线程池去执行。
    • submit()：是ExecutorService中声明的方法，在AbstractExecutorService就已经由了具体的实现，在ThreadPoolExecutor中并没有对其进行重写，这个方法是用来向线程池中提交任务的，但是它和execute()方法不同，它能够返回任务执行的结果，去看submit()方法的实现，会发现它实际上还是调用execute()方法，只不过他利用了Future来获取任务的执行结果。
    • shutdown()和shutdownNow()是用来关闭线程池的。
    4.深入剖析线程池实现原理

    （1）线程池的状态

    volatile int runState;

static final int RUNNING = 0;

static final int SHUTDOWN = 1;

static final int stop = 2;

static final int TERMINATED = 3;

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    • runState表示当前线程池的状态，用volatile变量用来保证线程之间的可见性。

    • 当创建线程池后，初始化时为RUNNING状态。

    • 如果调用了shutdown()方法，则线程池处于SHUTDOWN状态，此时它不能够接受新的任务，它会等所有的任务执行完毕。

    • 如果调用了shutdownNow()方法，则线程池处于SHOP状态，此时线程池不能接受新的任务，并且去尝试终止正在运行的任务。

    • 当线程池处于SHUTDOWN或者STOP状态，并且所有工作线程已经很销毁，任务缓存队列已经清空或执行结束后，线程池被设置为TERMINATEDzhua状态

    在这里插入图片描述

    （2）任务的执行

    • ThreadPoolExecutor类的核心成员：
    private final BlockingQueue<Runnable> workQueue; //任务缓存队列，用来存放等待执行的任务

private final ReetrantLock mainLock  = new ReetrantLock(); //线程池的主要锁状态，对线程池的状态改变的核心锁

private final HashSet<Worker> workers = new HashSet<Worker>(); //用来存放工作集

private volatile long keepAliveTime; //线程存活时间

private volatile boolean allowCoreThreadTimeOut; //是否允许为核心线程设置存活时间

private volatile int corePoolSize; //核心池的大小（即线程池中线程数目大于这个参数时，提交的任务会放在任务缓存队列）

private volatile int maximumPoolSize; //线程池中最大能容忍的线程数

private volatile int poolSize; //线程池中当前线程数

private volatile RejectedExecutionHandler handler; //任务拒绝策略

private volatile ThreadFactory threadFactory; //线程工厂，用来创建线程

private int largestPoolSize; //用来记录线程池中曾经出现的最大线程数

private long completedTaskCount; //用来记录已经执行完毕的任务个数

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 在ThreadPoolExecutor类中，最核心的任务提交方法是execute()方法，虽然通过submit也可以提交任务，但是实际上submit方法里面最终调用的还是execute()方法，所以我们只需要研究execute()方法的实现原理即可：
    public void execute(Runnable command) {
    //首先先判断传入的任务是否为空，若是null，则抛出空指针异常；
    if (command == null)
        throw new NullPointerException();
    //如果当前线程数量不小于核心池的数量或者执行addIfUnderCorePoolSize()方法，返回false主席那个下面代码块
    if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
        //判断线程池状态是否为RUNNING，并且放入缓存队列
        if (runState == RUNNING && workQueue.offer(command)) {
            //这句话是为了防止在将此任务添加进任务缓存队列的同时其他线程突然调用了shutdown或者shutdownNow方法时，那就调用
            //ensureQueuedTaskHandled方法
            if (runState != RUNNING || poolSize == 0)
                ensureQueuedTaskHandled(command);
        }
        //如果不是RUNNING状态，并且调用addIfUnderMaximumPoolSize方法失败，则执行拒绝处理。
        else if (!addIfUnderMaximumPoolSize(command))
            //拒绝处理
            reject(command); // is shutdown or saturated
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19

    （3）线程池中的线程初始化

    • 默认情况下，创建线程池之后，线程池中是没有线程的，需要提交任务后才会创建线程。

    • 在实际中如果需要线程池创建之后立即创建线程，可以通过以下两个办法办到：

      • prestartCoreThread()：初始化一个核心线程
      • prestartAllCoreThreads()：初始化所有核心线程
    public boolean prestartCoreThread(){
    return addIfUnderCorePoolSize(null); //注意传进去的参数是null
}

public int prestartAllCoreThreads(){
    int n = 0;
    while(addIfUnderCorePoolSize(null)) //注意传进去的参数是null
        ++n;
    return n;
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 注意上面传进去的参数null，r = workQueue.take()，即等待任务队列中有任务。

    （4）任务缓存队列以及排队策略

    • workQueue的类型为BlockingQueue ,通常可以取下面三种类型：
      • ArrayBlockingQueue：基于数组的先进先出队列，此队列创建时必须指定大小。
      • LinkedBlockingQueue：基于链表的先进先出队列，如果创建时没有指定此队列大小，则默认为Integer.MAX_VALUE。
      • synchronousQueue：这个队列比较特殊，它不会保存提交的任务，而是将直接新建一个线程来执行新来的任务。

    （5）任务拒绝策略

    • 当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize，如果还有任务到来就会采取任务拒绝策略，通常以下四种策略：
    ThreadPoolExecutor.AbortPolicy：丢弃任务并抛出RejectedExecutionException异常。
ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

    • 1
    • 2
    • 3
    • 4

    （6）线程池的关闭

    ThreadPoolExecutor提供了两个方法，用于线程池的关闭，分别是shutdown()和shutdownNow()，其中：

    • shutdown()：不会立即终止线程池，而是要等所有任务缓存队列中的任务都执行完后才终止，但再也不会接受新的任务
    • shutdownNow()：立即终止线程池，并尝试打断正在执行的任务，并且清空任务缓存队列，返回尚未执行的任务

    （7）线程池容量的动态调整

    ​ ThreadPoolExecutor提供了动态调整线程池容量大小的方法：setCorePoolSize()和setMaximumPoolSize()

    • setCorePoolSize：设置核心池大小
    • setMaximumPoolSize：设置线程池最大能创建的线程数目大小
    • 当上述参数从小变大时，ThreadPoolExecutor进行线程赋值，还可能立即创建新的线程来执行任务
    5.线程池使用示例
    public class ThreadPoolExecutorDemo {

    public static void main(String[] args) {
        //创建线程池，核心池5个，最大的线程池数量10个，多余线程空闲存活时间，任务缓存对列及排队策略
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(5));

        for (int i = 0; i < 15; i++) {
            //创建15个任务
            MyTask myTask = new MyTask(i);
            //每创建一个放在线程池中
            executor.execute(myTask);
            /**
             * getPoolSize():获取线程池中线程数目
             * getQueue().size():获取队列中等待的任务数目
             * getCompletedTaskCount():获取已经执行完成的任务数目
             */
            System.out.println("线程池中线程数目：" + executor.getPoolSize() + "，队列中等待执行的任务数目：" +
                    executor.getQueue().size() + "，已执行完的任务数目：" + executor.getCompletedTaskCount());
        }
        //结束线程池生命周期
        executor.shutdown();
    }
}

class MyTask implements Runnable {

    private int taskNum;

    public MyTask(int taskNum) {
        this.taskNum = taskNum;
    }

    @Override
    public void run() {
        System.out.println("正在执行task：" + taskNum);
        try {
            Thread.currentThread().sleep(4000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("task：" + taskNum + "执行完毕");
    }
}

    • 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

    在这里插入图片描述

    在这里插入图片描述

  • 相关阅读:
    10. awd 层(数据集市层)
    【SQL】MySQL中的存储引擎、事务、锁、日志
    132 分割回文串II
    使用VS Code 搭建 platformio 平台
    GitHub学生认证，可以愉快的使用Copilot了
    Aruba CX交换机VSX配置
    基于C++的简单ANN（人工神经网络）模型
    FRP内网穿透教程
    Codeforces Round #835 (Div. 4) - B. Atilla‘s Favorite Problem
    漫谈广告机制设计 | 混排：广告与自然结果的交锋博弈（2）
  • 原文地址：https://blog.csdn.net/weixin_47533244/article/details/127812088