线程池核心原理浅析

前言

由于系统资源是有限的，为了降低资源消耗，提高系统的性能和稳定性，引入了线程池对线程进行统一的管理和监控，本文将详细讲解线程池的使用、原理。

为什么使用线程池

池化思想

线程池主要用到了池化思想，池化思想在计算机领域十分常见，主要用于减少资源浪费、提高性能等。

池化思想主要包含以下几个方面：

一些常见的资源池包括线程池、数据库连接池、对象池、缓存池、连接池等。

池化思想可以提高系统的性能，因为它减少了资源的创建和销毁次数，避免了不必要的开销。通过池化，系统可以更好地应对高并发情况，降低资源竞争，提高响应速度。

什么是线程池

根据池化思想，在一个系统中，为了避免线程频繁的创建和销毁，让线程可以复用，引入了线程池的概念。线程池中，总有那么几个活跃线程。

当你需要使用线程时，可以从池子中随便拿一个空闲线程，当完成工作时，并不急着关闭线程，而是将这个线程退回到池子，方便其他人使用。

简单说就是，在使用线程池后，创建线程变成了从线程池中获得空闲线程，关闭线程编程了向池子里归还线程。

大致流程如下：

## 为什么使用线程池 Java 中的线程池是运用场景最多的并发框架，几乎所有需要异步或并发执行任务的程序都可以使用线程池。

在开发过程中，合理地使用线程池能够带来3个好处。

降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。
提高线程的可管理性。线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一分配、调优和监控。

要做到合理利用线程池，必须对其实现原理了如指掌。

线程池的使用

## ThreadPoolExecutor ThreadPoolExecutor 的创建方法总体来说可分为 2 种：

通过 ThreadPoolExecutor 构造函数
通过 Executors 类创建

通过构造函数

1.1. 入参含义

这个也是推荐使用的方法，因为通过 Executors 类创建可能会导致 OOM，如下图阿里开发规范中的描述。

构造函数入参：

 public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)

构造函数入参含义：

1.2. 阻塞队列

workQueue 可选的 BlockingQueue：

1.3. 拒绝策略

如下图，上述拒绝策略均实现 RejectedExecutionHandler 接口，且为 ThreadPoolExecutor 的内部类。

若以上策略仍无法满足实际应用需要，完全可以自已扩展 RejectedExecutionHandler 接口。

 public interface RejectedExecutionHandler {
 
    /**
     * @param r 当前请求执行的任务
     * @param executor 当前的线程池
     */
    void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

示例：

 public class RejectedExecutionDemo {
    public static class MyTask implements Runnable{
 
        @Override
        public void run() {
            System.out.println(new Date() + "：Thread ID is" + Thread.currentThread().getId());
 
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }
 
    public static void main(String[] args) throws InterruptedException {
        MyTask myTask = new MyTask();
        ExecutorService executorService = new ThreadPoolExecutor(5, 5,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(10),
                Executors.defaultThreadFactory(),
                (r, executor) -> System.out.println(r.hashCode() + "is discard")
        );
 
        for (int i = 0; i < 100; i++) {
            executorService.submit(myTask);
            Thread.sleep(10);
        }
    }
}

上述示例中，mytask 执行需要花费100毫秒，因此，必然会导致一些任务被直接丢弃。在实际应用中，我们可以将更详细的信息记录到日志中，来分析任务丢失情况和系统负载。

通过 Executors

Executors 类扮演着线程池工厂的角色，通过该类可以取得一个拥有定功能的线程池。

该类可以创建三种类型的 ThreadPoolExecutor：

FixedThreadPool
SingleThreadExecutor
CachedThreadPool

2.1. FixedThreadPool

固定线程数的线程池，该线程池中的线程数量始终不变。当有一个新的任务提交时，线程池中若有空闲线程，则立即执行。若没有，则新的任务会被暂时存在任务队列中，待有线程空闲时，在处理队列中的任务。

FixedThreadPool 使用的无界任务队列 LinkedBlockingQueue，可能造成内存泄露。

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

2.2. SingleThreadExecutor

只有一个工作线程的线程池，当多于 1 个任务被提交时，会存到任务队列中。该线程池使用的无界任务队列 LinkedBlockingQueue，可能造成内存泄露。

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

2.3. CachedThreadPool

根据实际情况调整线程数的线程池，线程池的线程数量不确定，若有空闲线程可复用，则会优先使用。若所有线程均在工作，此时新的任务则会创建新的线程优先处理。所有线程在任务执行完毕后，将返回线程池进行复用。

corePoolSize 被设置为0，maximumPoolSize 被设置为无界，存活时间设置为 60s，空闲线程超过60秒后将会被
终止。极端情况线程创建过多，会导致内存泄露。

 public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue());
}
 
 
public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue(),
                                  threadFactory);
}

ScheduledThreadPoolExecutor

简介

如下图， ScheduledThreadPoolExecutor 继承自ThreadPoolExecutor，它主要用来定期执行任务，功能与 Timer 类似且更加强大，可以在构造函数中指定多个对应的后台线程数。

使用

可通过 Executors 创建，源码如下：

 public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) {
    return new DelegatedScheduledExecutorService
        (new ScheduledThreadPoolExecutor(1, threadFactory));
}
 
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}
 
public static ScheduledExecutorService newScheduledThreadPool(
        int corePoolSize, ThreadFactory threadFactory) {
    return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}

这里的返回值是 ScheduledExecutorService，根据时间对线程进行调度。有三个主要方法：

 public interface ScheduledExecutorService extends ExecutorService {
 
    /**
     * 给定时间对任务进行调度
     */
    public ScheduledFuture schedule(Runnable command,
                                       long delay, TimeUnit unit);
 
    /**
     * 周期性对任务进行调度
     * 以第一个任务的开始时间 initialDelay + period 
     * 第一个任务在 initialDelay + period 执行
     * 第二个任务在 initialDelay + period * 2 执行
     */
    public ScheduledFuture scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit);
 
    /**
     * 周期性对任务进行调度
     * 上一个任务结束后，再经过 period 时间开始执行
     */
    public ScheduledFuture scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit);
}

如果任务遇到异常，那么后续的所有子任务都会停止调度，因此，必须保证异常被及时处理，为周期性任务的稳定调度提供条件。

ForkJoinPool

fork 是开启子进程，join 是等待，意思是分支子进程结束后才能得到结果，实际开发中，若频繁的 fork 开启线程可能严重影响系统性能，所以引入了 ForkJoinPool。

大致流程是，向 ForkJoinPool 线程池中提交一个 ForkJoinTask 任务，就是将任务分解成多个小任务，等任务全部完成后进行处理，这里采用了分治的思想，具体我将在后续单独展开，这里不多做赘述。

ForkJoin 可能出现两个问题：

子线程积累过多，可能导致系统性能严重下降；
调用层次过深，可能导致栈溢出。

线程池的任务提交

execute()

该方法用于提交不需要返回值的任务，且无法判断任务是否被线程池执行成功。

源码见下面的线程池原理章节。

submit()

该方法用于提交需要返回值的任务。线程池会返回 Future 对象，可以判断任务是否执行成功，还可以通过 Future 的get()方法来获取返回值。

get()方法会阻塞当前线程直到任务完成，还可以设置超时时间，到时立即返回，不过这时有可能任务没有执行完。

 public Future submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}

线程池的关闭

可以通过调用线程池的 shutdown 或 shutdownNow 方法来关闭线程池。

它们的原理是遍历线程池中的工作线程，然后逐个调用线程的 interrupt() 来中断线程，所以无法响应中断的任务可能永远无法终止。

两种方法存在一定的区别，shutdownNow首先将线程池的状态设置成 STOP，然后尝试停止所有的正在执行或暂停任务的线程，并返回等待执行任务的列表。而 shutdown 只是将线程池的状态设置成 SHUTDOWN 状态，然后中断所有没有正在执行任务的线程。

只要调用了这两个关闭方法中的任意一个，isShutdown方法就会返回true。当所有的任务都已关闭后，表示线程池关闭成功，这时调用isTerminaed方法会返回true。

至于应该调用哪一种方法来关闭线程池，应该由提交到线程池的任务特性决定，通常调用 shutdown 方法来关闭线程池，如果任务不一定要执行完，则可以调用 shutdownNow 方法。

线程池执行原理

执行源码

 public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
 
    int c = ctl.get();
    
    // 如果当前工作线程数是否小于核心线程数
    if (workerCountOf(c) < corePoolSize) {
        // 添加核心线程去执行任务，成功则return
        if (addWorker(command, true))
            return;
        // 添加失败，ctl有变化，需重新获取
        c = ctl.get();
    }
 
 
    // 判断是否为RUNNING，此时核心线程数已满，需加入任务队列
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        // 检查若不是RUNNING则将任务从队列移除
        if (! isRunning(recheck) && remove(command))
            // 执行拒绝策略
            reject(command);
            
        // 正常则添加一个非核心空线程，执行队列中的任务
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
 
    // 表示核心线程满了，队列也满了，创建非核心线程，执行任务
    else if (!addWorker(command, false))
        // 最大线程数也满了，走拒绝策略
        reject(command);
}

流程图

参考：
[1] 魏鹏. Java并发编程的艺术.
[2] 葛一鸣/郭超. 实战Java高并发程序设计.

相关阅读:
Spring是如何管理事务的？@Transactional注解详解
 使用全局配置处理字段名和属性名不一致的情况
 计算机网络第2章-HTTP和Web协议（2）
[网络工程师]-防火墙-入侵防护系统IPS
keil5MDK 及cubeMX安装教程
 Linux——进程地址空间
 〖Python 数据库开发实战 - MySQL篇㉗〗- MySQL 数字函数
 基于NPOI封装导出Excel方法
 如何写好一篇学术论文
 文件上传与下载
原文地址：https://www.cnblogs.com/fuxing/p/18180613

	public ThreadPoolExecutor(int corePoolSize,
	int maximumPoolSize,
	long keepAliveTime,
	TimeUnit unit,
	BlockingQueue workQueue,
	ThreadFactory threadFactory,
	RejectedExecutionHandler handler)

	public interface RejectedExecutionHandler {

	/**
	* @param r 当前请求执行的任务
	* @param executor 当前的线程池
	*/
	void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
	}

	public class RejectedExecutionDemo {
	public static class MyTask implements Runnable{

	@Override
	public void run() {
	System.out.println(new Date() + "：Thread ID is" + Thread.currentThread().getId());

	try {
	Thread.sleep(100);
	} catch (InterruptedException e) {
	throw new RuntimeException(e);
	}
	}
	}

	public static void main(String[] args) throws InterruptedException {
	MyTask myTask = new MyTask();
	ExecutorService executorService = new ThreadPoolExecutor(5, 5,
	0L, TimeUnit.MILLISECONDS,
	new LinkedBlockingQueue<>(10),
	Executors.defaultThreadFactory(),
	(r, executor) -> System.out.println(r.hashCode() + "is discard")
	);

	for (int i = 0; i < 100; i++) {
	executorService.submit(myTask);
	Thread.sleep(10);
	}
	}
	}

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

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

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

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

	public static ExecutorService newCachedThreadPool() {
	return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
	60L, TimeUnit.SECONDS,
	new SynchronousQueue());
	}


	public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
	return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
	60L, TimeUnit.SECONDS,
	new SynchronousQueue(),
	threadFactory);
	}

	public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) {
	return new DelegatedScheduledExecutorService
	(new ScheduledThreadPoolExecutor(1, threadFactory));
	}

	public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
	return new ScheduledThreadPoolExecutor(corePoolSize);
	}

	public static ScheduledExecutorService newScheduledThreadPool(
	int corePoolSize, ThreadFactory threadFactory) {
	return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
	}

	public interface ScheduledExecutorService extends ExecutorService {

	/**
	* 给定时间对任务进行调度
	*/
	public ScheduledFuture schedule(Runnable command,
	long delay, TimeUnit unit);

	/**
	* 周期性对任务进行调度
	* 以第一个任务的开始时间 initialDelay + period
	* 第一个任务在 initialDelay + period 执行
	* 第二个任务在 initialDelay + period * 2 执行
	*/
	public ScheduledFuture scheduleAtFixedRate(Runnable command,
	long initialDelay,
	long period,
	TimeUnit unit);

	/**
	* 周期性对任务进行调度
	* 上一个任务结束后，再经过 period 时间开始执行
	*/
	public ScheduledFuture scheduleWithFixedDelay(Runnable command,
	long initialDelay,
	long delay,
	TimeUnit unit);
	}

	public Future submit(Runnable task) {
	if (task == null) throw new NullPointerException();
	RunnableFuture ftask = newTaskFor(task, null);
	execute(ftask);
	return ftask;
	}

	public void execute(Runnable command) {
	if (command == null)
	throw new NullPointerException();

	int c = ctl.get();

	// 如果当前工作线程数是否小于核心线程数
	if (workerCountOf(c) < corePoolSize) {
	// 添加核心线程去执行任务，成功则return
	if (addWorker(command, true))
	return;
	// 添加失败，ctl有变化，需重新获取
	c = ctl.get();
	}


	// 判断是否为RUNNING，此时核心线程数已满，需加入任务队列
	if (isRunning(c) && workQueue.offer(command)) {
	int recheck = ctl.get();
	// 检查若不是RUNNING则将任务从队列移除
	if (! isRunning(recheck) && remove(command))
	// 执行拒绝策略
	reject(command);

	// 正常则添加一个非核心空线程，执行队列中的任务
	else if (workerCountOf(recheck) == 0)
	addWorker(null, false);
	}

	// 表示核心线程满了，队列也满了，创建非核心线程，执行任务
	else if (!addWorker(command, false))
	// 最大线程数也满了，走拒绝策略
	reject(command);
	}