1. Johngo学长首页
  2. 技术点详细复盘专题
  3. 人工智能

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

• 人工智能 • 阅读 69

文章目录

*
–
+
* 1.为什么要使用线程池
* 2.线程池创建线程
* 3.ThreadPoolExecutor类
* 4.深入剖析线程池实现原理
* 5.线程池使用示例

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

【并发编程】线程池及Executor框架
  • 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));
        }
    }
}

【并发编程】线程池及Executor框架
  • newScheduledThreadPool
  • 创建一个定长的线程池，支持定时周期性任务执行。

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

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

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

        for (int i = 0; i < 5; i++) {

            scheduledThreadPool.scheduleAtFixedRate(new ThreadForPools(i),2,3,TimeUnit.SECONDS);
        }
    }
}

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

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

        for (int i = 0; i < 5; i++) {

            scheduledThreadPool.scheduleWithFixedDelay(new ThreadForPools(i),5,3,TimeUnit.SECONDS);
        }
    }
}

【并发编程】线程池及Executor框架
  • 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));
        }
    }
}

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

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：由调用线程处理该任务
  • 类关系图

【并发编程】线程池及Executor框架 
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状态

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

（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) {

    if (command == null)
        throw new NullPointerException();

    if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {

        if (runState == RUNNING && workQueue.offer(command)) {

            if (runState != RUNNING || poolSize == 0)
                ensureQueuedTaskHandled(command);
        }

        else if (!addIfUnderMaximumPoolSize(command))

            reject(command);
    }
}

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

  • 默认情况下，创建线程池之后，线程池中是没有线程的，需要提交任务后才会创建线程。
  • 在实际中如果需要线程池创建之后立即创建线程，可以通过以下两个办法办到：
  • prestartCoreThread()：初始化一个核心线程
  • prestartAllCoreThreads()：初始化所有核心线程
public boolean prestartCoreThread(){
    return addIfUnderCorePoolSize(null);
}

public int prestartAllCoreThreads(){
    int n = 0;
    while(addIfUnderCorePoolSize(null))
        ++n;
    return n;
}
  • 注意上面传进去的参数null，r = workQueue.take()，即等待任务队列中有任务。

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

  • workQueue的类型为BlockingQueue

（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) {

        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(5));

        for (int i = 0; i < 15; i++) {

            MyTask myTask = new MyTask(i);

            executor.execute(myTask);

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

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

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

Original: https://blog.csdn.net/weixin_47533244/article/details/127812088
Author: 我就叫CV吧
Title: 【并发编程】线程池及Executor框架

原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/655908/

转载文章受原作者版权保护。转载请注明原作者出处！

赞 (0)
0

【自取】最近整理的，有需要可以领取学习：



Linux核心资料大放送~

全栈面试题汇总（持续更新&可下载）

一个提高学习100%效率的工具！

【超详细】深度学习面试题目！

LeetCode Python刷题答案下载！

LeetCode Java版刷题答案下载！

LeetCode C++ 版本，抓紧保存！

LeetCode GO语言 刷题答案下载！

大家都在看

亲爱的 Coder【最近整理，可免费获取】👉 最新必读书单  | 👏 面试题下载  | 🌎 免费的AI知识星球