本文目录导读：
我来详细介绍一下Java线程池的创建和使用方法,包含多个实际案例。
线程池的创建方法
1 使用Executors工厂类（不推荐生产环境）
import java.util.concurrent.*;
public class ThreadPoolExample1 {
public static void main(String[] args) {
// 创建固定线程数的线程池
ExecutorService fixedPool = Executors.newFixedThreadPool(3);
// 创建单线程线程池
ExecutorService singlePool = Executors.newSingleThreadExecutor();
// 创建缓存线程池
ExecutorService cachedPool = Executors.newCachedThreadPool();
// 创建定时任务线程池
ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(2);
// 提交任务
fixedPool.execute(() -> {
System.out.println("任务执行中: " + Thread.currentThread().getName());
});
// 关闭线程池
fixedPool.shutdown();
}
}
2 使用ThreadPoolExecutor（推荐方式）
public class ThreadPoolExample2 {
public static void main(String[] args) {
// 自定义线程池
ThreadPoolExecutor executor = new ThreadPoolExecutor(
2, // 核心线程数
4, // 最大线程数
60, // 空闲线程存活时间
TimeUnit.SECONDS, // 时间单位
new LinkedBlockingQueue<>(10), // 工作队列
Executors.defaultThreadFactory(), // 线程工厂
new ThreadPoolExecutor.AbortPolicy() // 拒绝策略
);
// 提交多个任务
for (int i = 0; i < 10; i++) {
final int taskId = i;
executor.execute(() -> {
System.out.println("任务" + taskId + " 由 " +
Thread.currentThread().getName() + " 执行");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
executor.shutdown();
}
}
完整案例：模拟银行柜员服务
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class BankServiceSimulation {
static class BankThreadFactory implements ThreadFactory {
private final AtomicInteger threadNumber = new AtomicInteger(1);
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r, "银行柜员-" + threadNumber.getAndIncrement());
t.setDaemon(false);
t.setPriority(Thread.NORM_PRIORITY);
return t;
}
}
public static void main(String[] args) {
// 创建银行服务线程池
ThreadPoolExecutor bankExecutor = new ThreadPoolExecutor(
2, // 核心柜员数
5, // 最大柜员数
30, // 空闲等待时间
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(10), // 等待队列容量
new BankThreadFactory(),
new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝时由调用线程处理
);
// 模拟客户到来
for (int i = 1; i <= 20; i++) {
final int customerId = i;
System.out.println("客户" + customerId + " 到达银行");
bankExecutor.execute(() -> {
System.out.println(Thread.currentThread().getName() +
" 正在服务客户" + customerId);
try {
// 模拟服务时间
Thread.sleep((long)(Math.random() * 3000));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println(Thread.currentThread().getName() +
" 完成客户" + customerId + "的服务");
});
try {
Thread.sleep(500); // 客户到达间隔
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// 监控线程池状态
monitorThreadPool(bankExecutor);
bankExecutor.shutdown();
}
private static void monitorThreadPool(ThreadPoolExecutor executor) {
ScheduledExecutorService monitor = Executors.newScheduledThreadPool(1);
monitor.scheduleAtFixedRate(() -> {
System.out.println("===== 线程池状态 =====");
System.out.println("活跃线程数: " + executor.getActiveCount());
System.out.println("完成任务数: " + executor.getCompletedTaskCount());
System.out.println("队列中任务数: " + executor.getQueue().size());
System.out.println("核心线程数: " + executor.getCorePoolSize());
System.out.println("最大线程数: " + executor.getMaximumPoolSize());
System.out.println("=====================");
}, 1, 2, TimeUnit.SECONDS);
// 5秒后停止监控
monitor.schedule(() -> {
monitor.shutdown();
}, 10, TimeUnit.SECONDS);
}
}
提交任务的不同方式
public class TaskSubmissionExample {
public static void main(String[] args) throws Exception {
ThreadPoolExecutor executor = new ThreadPoolExecutor(
2, 4, 60, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(100)
);
// 1. execute() - 提交无返回值的任务
executor.execute(() -> {
System.out.println("execute方式提交任务");
});
// 2. submit() - 提交有返回值的任务
Future<Integer> future1 = executor.submit(() -> {
Thread.sleep(1000);
return 42;
});
System.out.println("任务结果: " + future1.get());
// 3. submit() - 提交Callable任务
Future<String> future2 = executor.submit(new Callable<String>() {
@Override
public String call() throws Exception {
return "Callable任务结果";
}
});
System.out.println(future2.get());
// 4. invokeAll() - 提交多个任务
Callable<Integer> task1 = () -> 1;
Callable<Integer> task2 = () -> 2;
Callable<Integer> task3 = () -> 3;
List<Future<Integer>> futures = executor.invokeAll(
Arrays.asList(task1, task2, task3)
);
for (Future<Integer> future : futures) {
System.out.println("批量任务结果: " + future.get());
}
executor.shutdown();
}
}
实际应用：Web请求处理
import java.util.concurrent.*;
import java.util.*;
public class WebRequestHandler {
private final ThreadPoolExecutor requestPool;
private final Map<String, Long> requestStats = new ConcurrentHashMap<>();
public WebRequestHandler() {
this.requestPool = new ThreadPoolExecutor(
10, // 核心线程数
50, // 最大线程数
60,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(1000),
new ThreadPoolExecutor.CallerRunsPolicy()
);
}
public void handleRequest(String requestId, String url) {
long startTime = System.currentTimeMillis();
CompletableFuture.supplyAsync(() -> {
// 处理请求
System.out.println("处理请求 " + requestId + " URL: " + url);
try {
Thread.sleep(100); // 模拟处理时间
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return processUrl(url);
}, requestPool).thenAccept(result -> {
long duration = System.currentTimeMillis() - startTime;
requestStats.put(requestId, duration);
System.out.println("请求 " + requestId + " 处理完成，耗时: " + duration + "ms");
}).exceptionally(ex -> {
System.err.println("请求 " + requestId + " 处理失败: " + ex.getMessage());
return null;
});
}
private String processUrl(String url) {
// 模拟URL处理
return "Processed: " + url;
}
public void printStats() {
System.out.println("===== 请求统计 =====");
requestStats.forEach((id, duration) ->
System.out.println("请求 " + id + ": " + duration + "ms"));
}
public void shutdown() {
requestPool.shutdown();
try {
if (!requestPool.awaitTermination(30, TimeUnit.SECONDS)) {
requestPool.shutdownNow();
}
} catch (InterruptedException e) {
requestPool.shutdownNow();
}
}
public static void main(String[] args) {
WebRequestHandler handler = new WebRequestHandler();
// 模拟多个请求
for (int i = 0; i < 100; i++) {
handler.handleRequest("REQ-" + i,
"https://api.example.com/data/" + i);
}
// 等待所有请求完成
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
handler.printStats();
handler.shutdown();
}
}
线程池配置最佳实践
public class ThreadPoolBestPractices {
/**
* CPU密集型任务线程池
* 线程数 = CPU核心数 + 1
*/
public static ThreadPoolExecutor createCPUIntensivePool() {
int cpuCores = Runtime.getRuntime().availableProcessors();
return new ThreadPoolExecutor(
cpuCores + 1,
cpuCores + 1,
60, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(1000),
new ThreadPoolExecutor.CallerRunsPolicy()
);
}
/**
* IO密集型任务线程池
* 线程数 = CPU核心数 * 2
*/
public static ThreadPoolExecutor createIOIntensivePool() {
int cpuCores = Runtime.getRuntime().availableProcessors();
return new ThreadPoolExecutor(
cpuCores * 2,
cpuCores * 2,
60, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(2000),
new ThreadPoolExecutor.CallerRunsPolicy()
);
}
/**
* 混合型任务线程池
*/
public static ThreadPoolExecutor createMixedPool(int targetQueueSize) {
int cpuCores = Runtime.getRuntime().availableProcessors();
ThreadPoolExecutor executor = new ThreadPoolExecutor(
cpuCores,
cpuCores * 2,
30, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(targetQueueSize)
);
// 启用核心线程超时
executor.allowCoreThreadTimeOut(true);
return executor;
}
}
- 避免使用Executors创建线程池，推荐使用ThreadPoolExecutor手动配置
- 合理设置线程池参数，根据CPU密集/IO密集调整
- 选择合适的拒绝策略：
- AbortPolicy：抛出异常（默认）
- CallerRunsPolicy：调用线程执行
- DiscardPolicy：默默丢弃
- DiscardOldestPolicy：丢弃最旧任务
- 记得关闭线程池，使用shutdown()或shutdownNow()
- 监控线程池状态，及时调整参数
- 使用有界队列，避免OOM
这些案例覆盖了线程池的常见使用场景,你可以根据实际需求选择合适的方式。
