线程池的应用

左手的ㄟ右手 2023-09-28 23:52 41阅读 0赞

### 创建多线程的方式： ###

1.  继承Thread类创建线程类，重写run()方法
    
    1.  调用步骤，创建继承类的对象t,执行t.start()
2.  通过Runnable接口创建线程类，重写run()方法
    
    1.  调用步骤，创建实现类的对象r，创建Thread对象传入参数(r),通过start()方法启动
3.  通过Callable和Future创建线程，重写call()方法
    
    1.  创建线程池
    2.  创建接收结果的列表组合
    3.  创建线程对象
    4.  将线程对象提交到线程池中，并返回结果接收
    5.  将返回结果加入结果集合
    6.  关闭线程池
4.  基于线程池的execute()，创建临时线程
    
    1.  创建线程池
    2.  调用线程池的execute()方法
    3.  在用匿名内部类的方法创建Runnable对象，重写run()方法

### 线程池： ###

在执行一个异步任务或并发任务时，往往是通过直接new Thread()方法来创建新的线程，这样做弊端较多，更好的解决方案是合理地利用线程池，线程池的优势很明显，如下：

降低系统资源消耗，通过重用已存在的线程，降低线程创建和销毁造成的消耗；

提高系统响应速度，当有任务到达时，无需等待新线程的创建便能立即执行；

方便线程并发数的管控，线程若是无限制的创建，不仅会额外消耗大量系统资源，更是占用过多资源而阻塞系统或oom等状况，从而降低系统的稳定性。线程池能有效管控线程，统一分配、调优，提供资源使用率；

更强大的功能，线程池提供了定时、定期以及可控线程数等功能的线程池，使用方便简单。

java通过Executors提供四种线程池，分别为：

newCachedThreadPool：创建一个可缓存的无界线程池，如果线程池长度超过处理需要，可灵活回收空线程，若无可回收，则新建线程。当线程池中的线程空闲时间超过60s，则会自动回收该线程，当任务超过线程池的线程数则创建新的线程，线程池的大小上限为Integer.MAX\_VALUE,可看作无限大。

newFixedThreadPool：创建一个指定大小的线程池，可控制线程的最大并发数，超出的线程会在LinkedBlockingQueue阻塞队列中等待

newScheduledThreadPool：创建一个定长的线程池，可以指定线程池核心线程数，支持定时及周期性任务的执行

newSingleThreadExecutor：创建一个单线程化的线程池，它只有一个线程，用仅有的一个线程来执行任务，保证所有的任务按照指定顺序（FIFO，LIFO，优先级）执行，所有的任务都保存在队列LinkedBlockingQueue中，等待唯一的单线程来执行任务。

创建线程池，在构造一个新的线程池时，必须满足下面的条件：

**corePoolSize**（线程池基本大小）必须大于或等于0；当向线程池提交一个任务时，若线程池已创建的线程数小于corePoolSize，即便此时存在空闲线程，也会通过创建一个新线程来执行该任务，直到已创建的线程数大于或等于corePoolSize时，才会根据是否存在空闲线程，来决定是否需要创建新的线程。除了利用提交新任务来创建和启动线程（按需构造），也可以通过 prestartCoreThread() 或 prestartAllCoreThreads() 方法来提前启动线程池中的基本线程。

**maximumPoolSize**（线程池最大大小）必须大于或等于1，必须大于或等于corePoolSize；当向线程池提交一个任务时，若线程池已创建的线程数小于corePoolSize，即便此时存在空闲线程，也会通过创建一个新线程来执行该任务，直到已创建的线程数大于或等于corePoolSize时，才会根据是否存在空闲线程，来决定是否需要创建新的线程。除了利用提交新任务来创建和启动线程（按需构造），也可以通过 prestartCoreThread() 或 prestartAllCoreThreads() 方法来提前启动线程池中的基本线程。

**keepAliveTime**（线程存活保持时间）必须大于或等于0；默认情况下，当线程池的线程个数多于corePoolSize时，线程的空闲时间超过keepAliveTime则会终止。但只要keepAliveTime大于0，allowCoreThreadTimeOut(boolean) 方法也可将此超时策略应用于核心线程。另外，也可以使用setKeepAliveTime()动态地更改参数。

**workQueue**（任务队列）不能为空；用于传输和保存等待执行任务的阻塞队列。可以使用此队列与线程池进行交互：

如果运行的线程数少于 corePoolSize，则 Executor 始终首选添加新的线程，而不进行排队。

如果运行的线程数等于或多于 corePoolSize，则 Executor 始终首选将请求加入队列，而不添加新的线程。

如果无法将请求加入队列，则创建新的线程，除非创建此线程超出 maximumPoolSize，在这种情况下，任务将被拒绝。

**threadFactory**（线程工厂）不能为空，默认为DefaultThreadFactory类，用于创建新线程。由同一个threadFactory创建的线程，属于同一个ThreadGroup，创建的线程优先级都为Thread.NORM\_PRIORITY，以及是非守护进程状态。threadFactory创建的线程也是采用new Thread()方式，threadFactory创建的线程名都具有统一的风格：pool-m-thread-n（m为线程池的编号，n为线程池内的线程编号）;

**handler**（线程饱和策略）不能为空，默认策略为ThreadPoolExecutor.AbortPolicy。当线程池和队列都满了，则表明该线程池已达饱和状态。

ThreadPoolExecutor.AbortPolicy：处理程序遭到拒绝，则直接抛出运行时异常 RejectedExecutionException。(默认策略)

ThreadPoolExecutor.CallerRunsPolicy：调用者所在线程来运行该任务，此策略提供简单的反馈控制机制，能够减缓新任务的提交速度。

ThreadPoolExecutor.DiscardPolicy：无法执行的任务将被删除。

ThreadPoolExecutor.DiscardOldestPolicy：如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重新尝试执行任务（如果再次失败，则重复此过程）。

**unit**（存活时间的单位）：时间单位，分为7类，从细到粗顺序：NANOSECONDS（纳秒），MICROSECONDS（微妙），MILLISECONDS（毫秒），SECONDS（秒），MINUTES（分），HOURS（小时），DAYS（天）；

### **线程的corePoolSize、、maximumPoolSize、workQueue参数的作用和互相之间的关系？** ###

先说结论：corePoolSize-->workQueue-->maximumPoolSize

当n个请求过来的时候，线程池是这么处理的：

当n<=corePoolSize时：直接使用核心线程；

当n>corePoolSize时：先将核心线程占满，剩余的n-corePoolSize个请求将进入workQueue（阻塞队列），这里又分以下情况：

\--------当n-corePoolSize<=workQueue的最大容量，全部进入workQueue；

\--------当n-corePoolSize>workQueue的最大容量，请求先塞满workQueue，多余的请求n-corePoolSize-workQueue将根据maximumPoolSize-corePoolSize的余量开辟新的线程响应请求，

\--------直到达到maximumPoolSize，停止创建，超出的请求（n-maximumPoolSize-workQueue）直接被拒绝

来个demo：corePoolSize=3；maximumPoolSize=5；workQueue=5；

（1）当n<=3：直接占用核心线程池；

（2）当n=4~8：3个核心线程占满、1个请求进入workQueue；

（3）当n=9~10：3个核心线程占满、5个请求进入workQueue、剩下的请求新开线程响应；

（4）当n>10：3个核心线程占满、5个请求进入workQueue、2个请求被新开线程响应、其余的请求直接拒绝。

### 开发中如何创建线程池？ ###

1、构造方法实现：ThreadPollExecutor类

2、通过Executor框架的工具类Executors实现，可以创建三种类型的ThreadPoolExecutor：如newFixedThreadPool(int)、newSingleThreadExecutor()等

FixedThreadPool ： 该方法返回一个固定线程数量的线程池。该线程池中的线程数量始终不变。当有一个新的任务提交时，线程池中若有空闲线程，则立即执行。若没有，则新的任务会被暂存在一个任务队列中，待有线程空闲时，便处理在任务队列中的任务。  
SingleThreadExecutor： 方法返回一个只有一个线程的线程池。若多余一个任务被提交到该线程池，任务会被保存在一个任务队列中，待线程空闲，按先入先出的顺序执行队列中的任务。  
CachedThreadPool： 该方法返回一个可根据实际情况调整线程数量的线程池。线程池的线程数量不确定，但若有空闲线程可以复用，则会优先使用可复用的线程。若所有线程均在工作，又有新的任务提交，则会创建新的线程处理任务。所有线程在当前任务执行完毕后，将返回线程池进行复用。

### 线程池的开启与关闭 ###

1、开启

线程池ThreadPoolExecutor有四个构造方法

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 threadPoolExecutor = new ThreadPoolExecutor(1, 1, 10, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());
    threadPoolExecutor.execute(()->{
            // todo
        });

调用 execute() 方法，execute() 的源码：

public void execute(Runnable command) {
            if (command == null)
                throw new NullPointerException();
            int c = ctl.get();
            if (workerCountOf(c) < corePoolSize) { // 当前线程数是否大于核心线程数
                if (addWorker(command, true)) // 开启线程
                    return;
                c = ctl.get();
            }
            if (isRunning(c) && workQueue.offer(command)) {// 如果线程池处于running并且添加任务到等待队列成功
                int recheck = ctl.get();
                if (!isRunning(recheck) && remove(command)) // 如果线程池处于非running并且移除任务成功，那么驳回任务
                    reject(command);
                else if (workerCountOf(recheck) == 0) // 如果工作线程数为零，那么添加一个null的任务，以防新的任务进来却没有线程可以去工作
                    addWorker(null, false);
            } else if (!addWorker(command, false)) // 如果添加非核心线程时失败，那么驳回任务
                reject(command);
        }

execute() 方法的工作是先判断线程数是否大于核心线程数，小于就调用 addWorker() 方法添加核心线程，否则就添加到任务队列等待。在添加任务到队列采用 offer() 方法，如果返回 false ，那么就立即再次调用 addWorker() 方法添加一个非核心线程。

addWorker()方法：

private boolean addWorker(Runnable firstTask, boolean core) {
            retry:
            for (;;) {
                // 检验线程池...
            }
            boolean workerStarted = false;
            boolean workerAdded = false;
            Worker w = null; // 线程类
            try {
                w = new Worker(firstTask); // 把任务交给当前线程类
                final Thread t = w.thread;
                if (t != null) {
                    final ReentrantLock mainLock = this.mainLock;
                    mainLock.lock();
                    try {
                        int rs = runStateOf(ctl.get());
                        if (rs < SHUTDOWN ||
                                (rs == SHUTDOWN && firstTask == null)) {
                            if (t.isAlive())
                                throw new IllegalThreadStateException();
                            workers.add(w); // 把线程添加到HashSet中保存
                            int s = workers.size();
                            if (s > largestPoolSize)
                                largestPoolSize = s;
                            workerAdded = true;
                        }
                    } finally {
                        mainLock.unlock();
                    }
                    if (workerAdded) {
                        t.start(); // 开启线程
                        workerStarted = true;
                    }
                }
            } finally {
                // ...
            }
            return workerStarted;
        }

addWorker() 方法中，线程池的是把线程类封装成了一个任务，放到 Worker 类中执行。至此一个线程就开启成功了。

2、线程池工作

查看 Worker 这个类的 run() ，Worker 是线程池的任务执行类，实现了 Runnable 接口。Worker 类的 run 方法调用的是 runWorker() 这个方法。

final void runWorker(Worker w) {
            // ...
            Runnable task = w.firstTask; // 获取咱们提交的任务
            // ...
            try {
                while (task != null || (task = getTask()) != null) { // 获取任务去执行
                    // ...
                    try {
                        beforeExecute(wt, task);
                        Throwable thrown = null;
                        try {
                            task.run(); // 执行提交的任务
                        } catch (RuntimeException x) {
                            thrown = x; throw x;
                        } catch (Error x) {
                            thrown = x; throw x;
                        } catch (Throwable x) {
                            thrown = x; throw new Error(x);
                        } finally {
                            afterExecute(task, thrown);
                        }
                    } finally {
                        // ...
                    }
                }
                completedAbruptly = false;
            } finally {
                processWorkerExit(w, completedAbruptly);
            }
        }

runWorker() 拿到当前线程的任务，调用 task.run() 来执行业务侧提交的任务。处理完毕后，再次去等待队列中提取任务进行处理，如果没有任务那么终止这个 while 死循环，最后该线程结束。由于线程池的核心线程数是永久活跃的，那么这个 while() 循环就不能退出，否则该核心线程就结束了。查看getTask() 这个方法。

private Runnable getTask() {
            boolean timedOut = false; // Did the last poll() time out?
            for (;;) {
                // ...
                int wc = workerCountOf(c); // 工作线程数
                boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; // true-工作线程数>核心线程或者允许超时
                // ...
                try {
                    Runnable r = timed ?
                            workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                            workQueue.take(); // 如果允许超时，那么在 keepAliveTime 内没有 poll  获取数据则返回 null ，否则使用 take 方法会一直阻塞，直到获取新的数据。
                    if (r != null)
                        return r;
                    timedOut = true;
                } catch (InterruptedException retry) {
                    timedOut = false;
                }
            }
        }

allowCoreThreadTimeOut 这个参数，源码中的注释：如果为false(默认值)，核心线程即使在空闲时也保持活动。如果为true，核心线程使用keepAliveTime超时等待工作。

在默认情况下，allowCoreThreadTimeOut 是 false ，即不允许超时。再结合从队列获取任务的代码，对线程是怎么一直保持活跃的就十分清晰了。如果工作线程数大于核心线程数，那么在 keepAliveTime 中没有拉取到任务的话，就会返回一个 null ，这个 null 会使 runWorker() 的while循环终止，那么该非核心线程就运行完毕结束了。而如果工作线程数小于核心线程数，那么会使用 take() 方法一直阻塞的去拉取任务，直到新的任务添加进来，然后返回 runWorker() 去处理任务。

3、线程池关闭

ThreadPoolExecutor 为我们提供两种方式去关闭线程池：

①shutdown() 启动一个先前提交的有序关闭执行任务，但不接受新的任务。如果已经关闭，则调用没有额外的效果

②shutdownNow() 试图停止所有正在执行的任务，将停止处理等待中的任务，并返回任务列表等待处决的人这些任务被耗尽(删除)从此方法返回时从任务队列中返回。

shutdown() 调用后不再接受新的任务，并在当前任务执行完成后关闭线程池。shutdownNow() 调用后不再接受新的任务，并将当前等待队列中的任务返回给调用者，然后试图停止所有正在执行的任务，对于那种会抛出 InterruptedException 异常的方法会立即抛出异常。如果需要检测线程池是否真正关闭，使用 awaitTermination() 方法去查看线程状态。

两个方法的区别：

1.shutdown() 设置线程池的状态为 SHUTDOWN，shutdownNow() 设置线程池的状态为 STOP

2.shutdown() 什么都不返回。shutdownNow() 会把所有任务队列中的任务取出来，返回一个任务列表。

3.shutdown() 会执行完等待队列中的任务，shutdownNow() 不会。