Java线程池学习
线程池是Java并发编程中的一个重要工具，它能够有效地管理和控制线程的生命周期，从而提高程序的性能和响应能力。
一、线程池基础
1 定义
线程池是一种多线程处理形式，它通过预先创建一定数量的线程并将它们组织成一个池，来重用已存在的线程处理新提交的任务。这种做法可以避免频繁创建和销毁线程所带来的开销，同时还能有效控制并发线程的数量，避免过多线程消耗过多资源导致系统崩溃。
2 优点
- 资源复用：避免了频繁创建和销毁线程的开销。
- 控制最大并发数：限制系统中并发线程的数量，防止过度消耗系统资源。
- 提高响应速度：线程池中的线程通常处于等待状态，一旦有新任务提交，就可以立即处理。
- 方便管理线程：统一管理线程的创建、分配和回收过程。
3 基本组件
- 核心线程数 (
corePoolSize)：线程池中保持的最小线程数量。 - 最大线程数 (
maximumPoolSize)：线程池允许的最大线程数量。 - 空闲线程存活时间 (
keepAliveTime)：当线程池中的线程数量超过corePoolSize时，多余的空闲线程等待新任务的最长时间。 - 任务队列 (
workQueue)：用来存储等待执行的任务的阻塞队列。 - 线程工厂 (
threadFactory)：用于创建新线程的对象。 - 拒绝策略 (
handler)：当任务太多而无法被处理时采取的策略。
二、Java线程池实现
1 Executor接口
Executor接口定义了一个方法execute(Runnable command)，用于执行给定的任务。为了提供更丰富的功能，Java提供了ExecutorService接口作为扩展。
2 ExecutorService接口
ExecutorService接口继承自Executor，它增加了更多管理线程的方法，如shutdown()、isShutdown()、isTerminated()等。
3 ThreadPoolExecutor类
ThreadPoolExecutor是Java中最常用的线程池实现类，它提供了丰富的配置选项和控制机制。
3.1 创建线程池实例
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.TimeUnit;
public class ThreadPoolExample {
public static void main(String[] args) {
// 核心线程数
int corePoolSize = 5;
// 最大线程数
int maximumPoolSize = 10;
// 空闲线程存活时间
long keepAliveTime = 2L;
// 时间单位
TimeUnit unit = TimeUnit.MINUTES;
// 工作队列
LinkedBlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>();
// 线程工厂
ThreadFactory threadFactory = r -> new Thread(r);
// 拒绝策略
RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy();
// 创建线程池
ThreadPoolExecutor executor = new ThreadPoolExecutor(
corePoolSize,
maximumPoolSize,
keepAliveTime,
unit,
workQueue,
threadFactory,
handler
);
// 提交任务
for (int i = 0; i < 15; i++) {
Runnable worker = new WorkerThread("Task " + i);
executor.execute(worker);
}
// 关闭线程池
executor.shutdown();
while (!executor.isTerminated()) {
}
System.out.println("Finished all threads");
}
static class WorkerThread implements Runnable {
private String command;
public WorkerThread(String s) {
this.command = s;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " Start. Command = " + command);
processCommand();
System.out.println(Thread.currentThread().getName() + " End.");
}
private void processCommand() {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public String toString() {
return this.command;
}
}
}
三、执行策略
Java提供了几种预定义的执行策略，这些策略可以简化线程池的创建过程。
1 直接提交策略 (DirectSubmissionExecutor)
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class DirectSubmissionExample {
public static void main(String[] args) {
ExecutorService executor = Executors.newSingleThreadExecutor();
for (int i = 0; i < 10; i++) {
Runnable worker = new WorkerThread("Task " + i);
executor.execute(worker);
}
executor.shutdown();
while (!executor.isTerminated()) {
}
System.out.println("Finished all threads");
}
static class WorkerThread implements Runnable {
private String command;
public WorkerThread(String s) {
this.command = s;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " Start. Command = " + command);
processCommand();
System.out.println(Thread.currentThread().getName() + " End.");
}
private void processCommand() {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public String toString() {
return this.command;
}
}
}
2 固定线程数策略 (FixedThreadPool)
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class FixedThreadPoolExample {
public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(5);
for (int i = 0; i < 10; i++) {
Runnable worker = new WorkerThread("Task " + i);
executor.execute(worker);
}
executor.shutdown();
while (!executor.isTerminated()) {
}
System.out.println("Finished all threads");
}
static class WorkerThread implements Runnable {
private String command;
public WorkerThread(String s) {
this.command = s;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " Start. Command = " + command);
processCommand();
System.out.println(Thread.currentThread().getName() + " End.");
}
private void processCommand() {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public String toString() {
return this.command;
}
}
}
3 缓存线程池策略 (CachedThreadPool)
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class CachedThreadPoolExample {
public static void main(String[] args) {
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
Runnable worker = new WorkerThread("Task " + i);
executor.execute(worker);
}
executor.shutdown();
while (!executor.isTerminated()) {
}
System.out.println("Finished all threads");
}
static class WorkerThread implements Runnable {
private String command;
public WorkerThread(String s) {
this.command = s;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " Start. Command = " + command);
processCommand();
System.out.println(Thread.currentThread().getName() + " End.");
}
private void processCommand() {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public String toString() {
return this.command;
}
}
}
4 定时任务线程池 (ScheduledThreadPool)
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ScheduledThreadPoolExample {
public static void main(String[] args) {
ScheduledExecutorService scheduledExecutor = Executors.newScheduledThreadPool(5);
// 定时执行任务
scheduledExecutor.scheduleAtFixedRate(() -> System.out.println("Executing task at " + System.currentTimeMillis()), 0, 5, TimeUnit.SECONDS);
// 延迟执行任务
scheduledExecutor.schedule(() -> System.out.println("Executing delayed task at " + System.currentTimeMillis()), 10, TimeUnit.SECONDS);
// 停止调度
try {
Thread.sleep(60000);
} catch (InterruptedException e) {
e.printStackTrace();
}
scheduledExecutor.shutdown();
}
}
四、拒绝策略
当任务队列已满且线程池中的线程数量达到最大值时，线程池将采取一定的拒绝策略。
1 AbortPolicy
抛出RejectedExecutionException异常。
2 CallerRunsPolicy
由调用者所在的线程来运行任务。
3 DiscardPolicy
丢弃任务但不抛出异常。
4 DiscardOldestPolicy
丢弃队列中最老的任务，并尝试再次提交新的任务。
5 自定义拒绝策略
import java.util.concurrent.*;
public class CustomRejectPolicyExample {
public static void main(String[] args) {
ExecutorService executor = new ThreadPoolExecutor(
2,
5,
10,
TimeUnit.SECONDS,
new ArrayBlockingQueue<>(3),
new CustomRejectedExecutionHandler()
);
for (int i = 0; i < 10; i++) {
Runnable worker = new WorkerThread("Task " + i);
executor.execute(worker);
}
executor.shutdown();
while (!executor.isTerminated()) {
}
System.out.println("Finished all threads");
}
static class CustomRejectedExecutionHandler implements RejectedExecutionHandler {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
System.out.println("Task " + r.toString() + " rejected from " +
executor.toString());
}
}
static class WorkerThread implements Runnable {
private String command;
public WorkerThread(String s) {
this.command = s;
}
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " Start. Command = " + command);
processCommand();
System.out.println(Thread.currentThread().getName() + " End.");
}
private void processCommand() {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public String toString() {
return this.command;
}
}
}
五、线程池的配置与管理
1 配置核心参数
合理设置线程池的核心参数非常重要，以确保线程池能够在不同负载下高效运行。
2 监控与管理
监控线程池的状态，包括活动线程数、队列长度等，有助于及时发现和解决问题。
3 关闭线程池
正确关闭线程池可以避免资源泄露等问题。
六、应用案例
1 Web服务器场景
在Web服务器中，线程池可以用来处理HTTP请求。每个HTTP请求都会被分配到一个线程池中的线程上进行处理。这可以提高服务器的响应能力和吞吐量。
示例代码
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class WebServerExample {
public static void main(String[] args) throws IOException {
// 创建固定大小的线程池
ExecutorService executor = Executors.newFixedThreadPool(50);
// 创建服务器监听端口
ServerSocket serverSocket = new ServerSocket(8080);
System.out.println("Server started on port 8080");
try {
while (true) {
Socket clientSocket = serverSocket.accept();
Runnable worker = new ClientHandler(clientSocket);
executor.execute(worker);
}
} finally {
serverSocket.close();
executor.shutdown();
}
}
static class ClientHandler implements Runnable {
private final Socket socket;
public ClientHandler(Socket socket) {
this.socket = socket;
}
@Override
public void run() {
try {
System.out.println("Handling request from " + socket.getInetAddress().getHostAddress());
// 处理客户端请求
// ...
} catch (IOException e) {
e.printStackTrace();
} finally {
try {
socket.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
2 批处理任务
利用线程池可以高效地处理大量任务，比如文件批量处理、数据计算等。
示例代码
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
public class BatchProcessingExample {
public static void main(String[] args) {
// 创建可缓存线程池
ExecutorService executor = Executors.newCachedThreadPool();
// 创建一批任务
List<Runnable> tasks = IntStream.range(0, 1000)
.boxed()
.map(i -> new Task("Task " + i))
.collect(Collectors.toList());
// 提交任务到线程池
tasks.forEach(executor::submit);
// 关闭线程池
executor.shutdown();
while (!executor.isTerminated()) {
}
System.out.println("Finished all tasks");
}
static class Task implements Runnable {
private String name;
public Task(String name) {
this.name = name;
}
@Override
public void run() {
System.out.println("Processing " + name);
try {
Thread.sleep(1000); // 模拟处理任务
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
3 异步处理
线程池在异步处理中非常有用，可以用来处理后台任务，避免阻塞主线程。
示例代码
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class AsyncProcessingExample {
public static void main(String[] args) {
// 创建固定大小的线程池
ExecutorService executor = Executors.newFixedThreadPool(10);
// 提交异步任务
for (int i = 0; i < 10; i++) {
Runnable worker = new AsyncWorker("Task " + i);
executor.execute(worker);
}
// 关闭线程池
executor.shutdown();
while (!executor.isTerminated()) {
}
System.out.println("Finished all async tasks");
}
static class AsyncWorker implements Runnable {
private String name;
public AsyncWorker(String name) {
this.name = name;
}
@Override
public void run() {
System.out.println("Starting " + name);
try {
Thread.sleep(1000); // 模拟异步处理
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Completed " + name);
}
}
}
七、性能考虑
1 吞吐量与延迟
不同的策略下，线程池的吞吐量和延迟表现会有所不同。
2 资源利用率
线程池可以提高系统的资源利用率，尤其是在高并发场景下。