线程池核心源码深度剖析：原理、实战与优化
原创
线程池核心源码深度剖析：原理、实战与优化
原创
一、背景知识
在当今的软件开发领域，尤其是对于需要处理高并发、多任务的系统，多线程编程已经成为一项必备技能。多线程能够显著提升系统性能，使程序能够充分利用多核处理器，实现并行处理，进而缩短任务执行时间。然而，直接使用线程来处理任务存在诸多问题，例如：
- 资源开销：创建和销毁线程会消耗大量系统资源，包括内存和 CPU 时间，因为每个线程都需要分配独立的栈空间和上下文切换开销。
- 资源管理困难：如果无节制地创建线程，可能会耗尽系统资源，导致系统性能下降甚至崩溃。
为了克服这些问题，线程池应运而生。线程池是一种管理和复用线程的机制，它提前创建一定数量的线程，将任务分配给这些线程执行，避免了频繁创建和销毁线程的开销，提高了系统的性能和资源利用率。
二、为什么项目中要采用线程池？
（一）资源复用
线程池允许预先创建一组线程，这些线程在完成任务后不会立即销毁，而是处于等待状态，当新任务到来时可以立即复用，避免了频繁创建和销毁线程的成本，提高了系统的响应速度和性能。
（二）资源管理与控制
通过线程池，可以对线程的数量进行管理和控制，避免因过多线程同时运行而导致的资源耗尽，同时也可以根据系统的负载情况动态调整线程数量，提高系统的稳定性和可靠性。
（三）任务调度
线程池可以根据任务的优先级和执行顺序，合理地分配线程资源，确保任务的有序执行，提高任务的执行效率。
三、Java 中自带的线程池有哪些？
（一）FixedThreadPool
Executors.newFixedThreadPool(int nThreads) 创建一个固定大小的线程池，该线程池包含固定数量的线程，即核心线程数等于最大线程数。
import java.time.LocalDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class FixedThreadPoolExample {
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(3);
System.out.println("执行开始时间"+ LocalDateTime.now());
for (int i = 0; i < 10; i++) {
executorService.execute(() -> {
System.out.println(Thread.currentThread().getName() + " is running");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
executorService.shutdown();
System.out.println("执行完成时间"+ LocalDateTime.now());
}
}执行结果，由于开的线程进行执行的，所以整个调用没有耗时，每个线程都是开的独立线程进行执行的；
代码解释：
- 此代码使用
Executors.newFixedThreadPool(3)创建了一个具有 3 个线程的线程池。 - 通过
execute()方法向线程池提交 10 个任务，这些任务会被分配给线程池中的线程执行。 - 每个任务打印当前线程名称并休眠 1 秒，模拟任务执行。
（二）CachedThreadPool
Executors.newCachedThreadPool() 创建一个可缓存的线程池，线程池的大小会根据需要自动调整，适合执行大量短期异步任务。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class CachedThreadPoolExample {
public static void main(String[] args) {
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
executorService.execute(() -> {
System.out.println(Thread.currentThread().getName() + " is running");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
executorService.shutdown();
}
}代码解释：
- 创建一个可缓存的线程池，线程数量根据任务数量动态调整。
- 任务执行完成后，如果线程空闲一段时间（默认为 60 秒），将被回收。
（三）SingleThreadExecutor
Executors.newSingleThreadExecutor() 创建一个单线程的线程池，保证所有任务按照提交顺序依次执行。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class SingleThreadExecutorExample {
public static void main(String[] args) {
ExecutorService executorService = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
executorService.execute(() -> {
System.out.println(Thread.currentThread().getName() + " is running");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
executorService.shutdown();
}
}代码解释：
- 此线程池只有一个线程，所有任务依次执行，适用于需要保证任务执行顺序的场景。
四、为何要自己手动创建线程池？
（一）定制化需求
Java 自带的线程池虽然方便，但在某些场景下无法满足定制化需求，例如需要自定义线程池的核心参数，如核心线程数、最大线程数、任务队列类型、拒绝策略等。
（二）资源控制与优化
使用 Executors 创建的线程池可能存在资源耗尽的风险，如 CachedThreadPool 可能会创建大量线程，导致系统资源耗尽。手动创建线程池可以更好地根据实际情况优化资源使用。
五、线程池的核心属性 ctl 的作用
（一）ctl 的定义
ctl 是一个 AtomicInteger 类型的变量，它是线程池状态和工作线程数量的组合表示，使用低 29 位表示线程数量，高 3 位表示线程池状态。
（二）位运算
通过位运算来同时管理线程池状态和线程数量，如 private static int runStateOf(int c) { return c & ~CAPACITY; } 用于获取线程池状态，private static int workerCountOf(int c) { return c & CAPACITY; } 用于获取线程数量。
六、线程池的状态有哪些？
（一）RUNNING
线程池处于运行状态，接受新任务并处理队列中的任务。
（二）SHUTDOWN
不再接受新任务，但会继续处理队列中的任务。
（三）STOP
不接受新任务，也不处理队列中的任务，会中断正在执行的任务。
（四）TIDYING
所有任务都已终止，即将进行清理工作。
（五）TERMINATED
线程池终止完成。
七、线程池的状态流转机制
（一）RUNNING -> SHUTDOWN
调用 shutdown() 方法时，线程池进入 SHUTDOWN 状态，不再接受新任务，但继续处理队列中的任务。
（二）RUNNING 或 SHUTDOWN -> STOP
调用 shutdownNow() 方法，线程池进入 STOP 状态，中断正在执行的任务，不处理队列中的任务。
（三）SHUTDOWN -> TIDYING
当队列中的任务和正在执行的任务都完成时，进入 TIDYING 状态。
（四）TIDYING -> TERMINATED
当 terminated() 方法执行完毕，进入 TERMINATED 状态。
八、线程池提交任务的方式有哪些？
（一）execute 方法
execute(Runnable command) 方法用于提交一个 Runnable 任务，如果无法执行会抛出 RejectedExecutionException。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ExecuteMethodExample {
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(2);
executorService.execute(() -> {
System.out.println("Task is running");
});
executorService.shutdown();
}
}代码解释：
- 使用
execute()方法将Runnable任务提交给线程池执行。
（二）submit 方法
submit(Callable<T> task) 方法可提交 Callable 任务，并返回 Future<T> 对象，可获取任务执行结果或取消任务。
import java.util.concurrent.*;
public class SubmitMethodExample {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(2);
Future<String> future = executorService.submit(() -> {
return "Task completed";
});
System.out.println(future.get());
executorService.shutdown();
}
}代码解释：
- 使用
submit()方法提交Callable任务，通过Future.get()获取任务结果。
九、线程池的 7 个参数有哪些？
（一）corePoolSize
核心线程数，即使空闲也不会被销毁的线程数量。
（二）maximumPoolSize
线程池允许的最大线程数。
（三）keepAliveTime
当线程数大于核心线程数时，多余线程的空闲时间超过此值将被销毁。
（四）unit
keepAliveTime 的时间单位。
（五）workQueue
存储等待执行任务的队列，如 ArrayBlockingQueue、LinkedBlockingQueue 等。
（六）threadFactory
创建新线程的工厂，可自定义线程的名称、优先级等属性。
（七）rejectedExecutionHandler
任务无法执行时的拒绝策略，如 AbortPolicy、CallerRunsPolicy 等。
十、线程池提交任务后的底层流程
（一）提交任务
当使用 execute() 或 submit() 方法提交任务时：
- 首先检查线程池状态，若为
RUNNING则继续。 - 若线程数小于
corePoolSize，创建新的核心线程执行任务。 - 若线程数大于等于
corePoolSize，将任务添加到workQueue。 - 若
workQueue已满且线程数小于maximumPoolSize，创建新的非核心线程执行任务。 - 若
workQueue已满且线程数达到maximumPoolSize，根据rejectedExecutionHandler处理任务。
十一、线程池提交任务到工作队列后的细节处理
（一）任务队列类型
- ArrayBlockingQueue：有界阻塞队列，基于数组实现，需要指定容量。
- LinkedBlockingQueue：基于链表的阻塞队列，可选择有界或无界。
- SynchronousQueue：不存储元素，直接将任务交给线程，适用于传递性任务。
（二）阻塞与非阻塞队列
- 阻塞队列会在队列满或空时阻塞线程，确保线程安全；非阻塞队列会直接返回结果，可能导致任务丢失或异常。
十二、Java 实战例子
示例一：自定义线程池及参数优化
import java.util.concurrent.*;
public class CustomThreadPoolExample {
public static void main(String[] args) {
ThreadPoolExecutor executor = new ThreadPoolExecutor(
2, // corePoolSize
4, // maximumPoolSize
60, TimeUnit.SECONDS, // keepAliveTime and unit
new LinkedBlockingQueue<>(10), // workQueue
new ThreadFactoryBuilder().setNameFormat("custom-thread-%d").build(), // threadFactory
new ThreadPoolExecutor.CallerRunsPolicy() // rejectedExecutionHandler
);
for (int i = 0; i < 20; i++) {
executor.execute(() -> {
System.out.println(Thread.currentThread().getName() + " is running");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
executor.shutdown();
}
}代码解释：
- 创建一个自定义线程池，核心线程数为 2，最大线程数为 4。
- 使用
ThreadFactoryBuilder自定义线程工厂，设置线程名称格式。 - 使用
CallerRunsPolicy作为拒绝策略，当任务无法执行时，由调用者线程执行任务。
示例二：基于线程池的异步任务处理
import java.util.concurrent.*;
public class AsyncTaskThreadPool {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(3);
CompletableFuture.runAsync(() -> {
System.out.println(Thread.currentThread().getName() + " is running");
}, executorService).thenRun(() -> {
System.out.println("Task completed");
});
executorService.shutdown();
}
}代码解释：
- 使用
CompletableFuture.runAsync()方法提交一个异步任务到线程池。 thenRun()方法在任务完成后执行后续操作。
示例三：线程池的动态调整与监控
import java.util.concurrent.*;
public class DynamicThreadPoolExample {
public static void main(String[] args) throws InterruptedException {
ThreadPoolExecutor executor = new ThreadPoolExecutor(
2, // corePoolSize
4, // maximumPoolSize
60, TimeUnit.SECONDS, // keepAliveTime and unit
new LinkedBlockingQueue<>(10) // workQueue
);
for (int i = 0; i < 20; i++) {
executor.execute(() -> {
System.out.println(Thread.currentThread().getName() + " is running");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
// 动态调整核心线程数
executor.setCorePoolSize(3);
executor.shutdown();
}
}代码解释：
- 创建一个线程池并提交任务。
- 使用
setCorePoolSize()方法动态调整核心线程数，可根据系统负载灵活调整线程池规模。
十三、性能、易扩展和稳定性的考虑
（一）性能
- 核心线程数和最大线程数：
- 对于 CPU 密集型任务，核心线程数可设置为
Runtime.getRuntime().availableProcessors()，避免过多线程导致的上下文切换开销。 - 对于 I/O 密集型任务，可适当增加核心线程数，以充分利用等待 I/O 的时间。
- 对于 CPU 密集型任务，核心线程数可设置为
- 任务队列选择：
- 有界队列可避免任务过多导致的资源耗尽，无界队列在任务生产速度大于消费速度时可能导致内存溢出。
（二）易扩展
- 动态调整：
- 可使用
setCorePoolSize()和setMaximumPoolSize()动态调整线程池规模。 - 利用自定义线程工厂，根据任务特点创建不同属性的线程，便于扩展。
- 可使用
（三）稳定性
- 异常处理：
- 在任务执行时，使用
try-catch处理异常，防止异常导致线程池异常终止。 - 选择合适的拒绝策略，避免任务丢失或系统崩溃。
- 在任务执行时，使用
十四、底层原理
（一）线程池的初始化
- 线程池在初始化时会创建核心线程，这些线程会等待任务到来，同时创建任务队列存储等待执行的任务。
（二）任务调度与执行
- 当任务提交时，根据线程池状态和线程数量进行调度，涉及到线程创建、任务入队和拒绝策略的执行。
- 工作线程不断从任务队列中获取任务执行，任务执行完后继续获取下一个任务，直到线程池关闭或队列为空。
（三）状态管理
- 通过
ctl属性和位运算管理线程池状态和线程数量，实现状态的安全转换和线程数量的准确控制。
十五、总结
线程池是 Java 并发编程中的重要组件，通过理解其核心源码和工作原理，我们可以更好地使用和优化线程池。在实际项目中，根据具体业务需求和系统性能，合理选择线程池类型，手动创建定制化的线程池，并对线程池进行性能优化、扩展和稳定性保障。作为大数据工程师，掌握线程池的原理和使用技巧，有助于开发高性能、高并发的系统，避免因多线程引发的性能和资源管理问题。通过上述实战示例，可以将理论知识应用于实践，根据不同场景灵活调整线程池，提升系统的整体性能和稳定性。
以上文章从多个维度对线程池进行了深入剖析，包含其背景、使用、源码原理、实战示例和性能优化，帮助你更好地理解和使用线程池，提升多线程编程的能力。在实际开发中，根据具体的业务需求和系统特点，深入考虑性能、扩展和稳定性因素，确保系统的高效运行。
请注意，在多线程编程中，需要特别关注线程安全问题，例如同步问题、资源竞争和死锁等，避免因并发问题影响系统性能和稳定性。同时，合理使用线程池可以显著提高系统性能，但需要谨慎设计和测试，确保线程池在不同负载下的表现符合预期。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
腾讯云开发者
