AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

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

2023-05-24 77
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 【并发编程】线程池及Executor框架

1.为什么要使用线程池

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

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

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

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();
        }
    }
}
  • 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));
        }
    }
}

52af2b9c78674bd6af19b690ecbaa2e4.jpg

  • 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));
        }
    }
}

7f4b8d70d67e43788e74d8d9ca9d45a8.jpg

  • 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);
        }
    }
}

47fc8d41f1494f30a5a065e575f917cb.jpg

//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);
        }
    }
}

cacd5edf17884ddcbd2a075456d2c754.jpg

//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);
        }
    }
}


76f4c45c98254a37a1c42c2798719764.jpg

  • 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));
        }
    }
}

b38cf69f26494d5894aeba30745ccb5f.jpg

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);
    ...
}

从上面的代码可以得知，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：由调用线程处理该任务

类关系图

484bad368e8f46e1aea9c7aed3c495c5.jpg

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

在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;

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

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

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

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

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



7a8228f0bc7640caa75b694ec396a9fb.jpg

（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; //用来记录已经执行完毕的任务个数
  • 在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
    }
}

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

  • 默认情况下，创建线程池之后，线程池中是没有线程的，需要提交任务后才会创建线程。
  • 在实际中如果需要线程池创建之后立即创建线程，可以通过以下两个办法办到：
  • prestartCoreThread()：初始化一个核心线程
  • prestartAllCoreThreads()：初始化所有核心线程
public boolean prestartCoreThread(){
    return addIfUnderCorePoolSize(null); //注意传进去的参数是null
}
public int prestartAllCoreThreads(){
    int n = 0;
    while(addIfUnderCorePoolSize(null)) //注意传进去的参数是null
        ++n;
    return n;
}
  • 注意上面传进去的参数null，r = workQueue.take()，即等待任务队列中有任务。

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

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

（5）任务拒绝策略

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

（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 + "执行完毕");
    }
}


8107dca37bb04d23a3d494030657f684.jpg

fa51698571e644a0af19790d85ff90bb.jpg

文章标签：
Java
缓存
存储
关键词：
并发编程线程
线程executor
线程框架
互联网小阿祥
目录
相关文章
土木林森
|
4月前
|
Java 程序员 调度
【JAVA 并发秘籍】进程、线程、协程：揭秘并发编程的终极武器！
【8月更文挑战第25天】本文以问答形式深入探讨了并发编程中的核心概念——进程、线程与协程，并详细介绍了它们在Java中的应用。文章不仅解释了每个概念的基本原理及其差异，还提供了实用的示例代码，帮助读者理解如何在Java环境中实现这些并发机制。无论你是希望提高编程技能的专业开发者，还是准备技术面试的求职者，都能从本文获得有价值的见解。
土木林森
66 1
叫个什么名字
|
17天前
|
并行计算 数据处理 调度
Python中的并发编程：探索多线程与多进程的奥秘####
本文深入探讨了Python中并发编程的两种主要方式——多线程与多进程，通过对比分析它们的工作原理、适用场景及性能差异，揭示了在不同应用需求下如何合理选择并发模型。文章首先简述了并发编程的基本概念，随后详细阐述了Python中多线程与多进程的实现机制，包括GIL（全局解释器锁）对多线程的影响以及多进程的独立内存空间特性。最后，通过实例演示了如何在Python项目中有效利用多线程和多进程提升程序性能。 ####
叫个什么名字
34 5
土木林森
|
4月前
|
Java 开发者
解锁并发编程新姿势！深度揭秘AQS独占锁&ReentrantLock重入锁奥秘，Condition条件变量让你玩转线程协作，秒变并发大神！
【8月更文挑战第4天】AQS是Java并发编程的核心框架，为锁和同步器提供基础结构。ReentrantLock基于AQS实现可重入互斥锁，比`synchronized`更灵活，支持可中断锁获取及超时控制。通过维护计数器实现锁的重入性。Condition接口允许ReentrantLock创建多个条件变量，支持细粒度线程协作，超越了传统`wait`/`notify`机制，助力开发者构建高效可靠的并发应用。
土木林森
91 0
萝卜带泥
|
2月前
|
数据挖掘 程序员 调度
探索Python的并发编程：线程与进程的实战应用
【10月更文挑战第4天】 本文深入探讨了Python中实现并发编程的两种主要方式——线程和进程，通过对比分析它们的特点、适用场景以及在实际编程中的应用，为读者提供清晰的指导。同时，文章还介绍了一些高级并发模型如协程，并给出了性能优化的建议。
萝卜带泥
31 3
丰宝宝
|
3月前
|
负载均衡 Java 调度
探索Python的并发编程：线程与进程的比较与应用
本文旨在深入探讨Python中的并发编程，重点比较线程与进程的异同、适用场景及实现方法。通过分析GIL对线程并发的影响，以及进程间通信的成本，我们将揭示何时选择线程或进程更为合理。同时，文章将提供实用的代码示例，帮助读者更好地理解并运用这些概念，以提升多任务处理的效率和性能。
丰宝宝
60 3
游客7ofzsdfcat7ls
|
3月前
|
缓存 监控 Java
Java中的并发编程：理解并应用线程池
在Java的并发编程中，线程池是提高应用程序性能的关键工具。本文将深入探讨如何有效利用线程池来管理资源、提升效率和简化代码结构。我们将从基础概念出发，逐步介绍线程池的配置、使用场景以及最佳实践，帮助开发者更好地掌握并发编程的核心技巧。
游客7ofzsdfcat7ls
39 10
游客moiomvrp3vyac2
|
3月前
|
并行计算 API 调度
探索Python中的并发编程：线程与进程的对比分析
【9月更文挑战第21天】本文深入探讨了Python中并发编程的核心概念，通过直观的代码示例和清晰的逻辑推理，引导读者理解线程与进程在解决并发问题时的不同应用场景。我们将从基础理论出发，逐步过渡到实际案例分析，旨在揭示Python并发模型的内在机制，并比较它们在执行效率、资源占用和适用场景方面的差异。文章不仅适合初学者构建并发编程的基础认识，同时也为有经验的开发者提供深度思考的视角。
游客moiomvrp3vyac2
28 0
做梦都在改BUG
|
4月前
|
数据采集 Java Python
Python并发编程：多线程（threading模块）
Python是一门强大的编程语言，提供了多种并发编程方式，其中多线程是非常重要的一种。本文将详细介绍Python的threading模块，包括其基本用法、线程同步、线程池等，最后附上一个综合详细的例子并输出运行结果。
做梦都在改BUG
86 7
做梦都在改BUG
|
4月前
|
数据采集 Java Python
Python并发编程：多线程（threading模块）
本文详细介绍了Python的threading模块，包括线程的创建、线程同步、线程池的使用，并通过多个示例展示了如何在实际项目中应用这些技术。通过学习这些内容，您应该能够熟练掌握Python中的多线程编程，提高编写并发程序的能力。 多线程编程可以显著提高程序的并发性能，但也带来了新的挑战和问题。在使用多线程时，需要注意避免死锁、限制共享资源的访问，并尽量使用线程池来管理和控制线程。
做梦都在改BUG
39 1
游客nmbcre4jd7shs
|
4月前
|
缓存 Java 数据处理
Java中的并发编程：解锁多线程的力量
在Java的世界里，并发编程是提升应用性能和响应能力的关键。本文将深入探讨Java的多线程机制，从基础概念到高级特性，逐步揭示如何有效利用并发来处理复杂任务。我们将一起探索线程的创建、同步、通信以及Java并发库中的工具类，带你领略并发编程的魅力。
游客nmbcre4jd7shs
39 2