Java线程池案例如何创建使用

本文目录导读：

1. 线程池的创建方法
2. 完整案例：模拟银行柜员服务
3. 提交任务的不同方式
4. 实际应用：Web请求处理
5. 线程池配置最佳实践

我来详细介绍一下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;
    }
}

1. 避免使用Executors创建线程池，推荐使用ThreadPoolExecutor手动配置
2. 合理设置线程池参数，根据CPU密集/IO密集调整
3. 选择合适的拒绝策略：
   • AbortPolicy：抛出异常（默认）
   • CallerRunsPolicy：调用线程执行
   • DiscardPolicy：默默丢弃
   • DiscardOldestPolicy：丢弃最旧任务
4. 记得关闭线程池，使用shutdown()或shutdownNow()
5. 监控线程池状态，及时调整参数
6. 使用有界队列，避免OOM

这些案例覆盖了线程池的常见使用场景,你可以根据实际需求选择合适的方式。