Java线程池的使用实例

发布时间：2021-08-21 23:56:17 作者：chen
来源：亿速云阅读：601

# Java线程池的使用实例

## 一、线程池概述

### 1.1 什么是线程池
线程池（Thread Pool）是一种多线程处理形式，处理过程中将任务添加到队列，然后在创建线程后自动启动这些任务。线程池中的线程可以重复利用，减少了频繁创建和销毁线程带来的性能开销。

### 1.2 为什么需要线程池
- **资源消耗控制**：减少线程创建/销毁的系统开销
- **响应速度提升**：任务到达时无需等待线程创建
- **线程管理**：提供线程限制、监控等功能
- **任务队列**：缓冲来不及处理的任务

### 1.3 Java线程池框架
Java通过`java.util.concurrent`包提供线程池支持，核心接口和类包括：
- `Executor`：执行任务的简单接口
- `ExecutorService`：扩展了Executor的功能
- `ThreadPoolExecutor`：可配置的线程池实现
- `Executors`：线程池工厂类

## 二、线程池核心参数

### 2.1 ThreadPoolExecutor构造参数
```java
public ThreadPoolExecutor(
    int corePoolSize,
    int maximumPoolSize,
    long keepAliveTime,
    TimeUnit unit,
    BlockingQueue<Runnable> workQueue,
    ThreadFactory threadFactory,
    RejectedExecutionHandler handler
)

2.2 参数详解

参数名	说明
corePoolSize	核心线程数，即使空闲也不会被回收
maximumPoolSize	线程池最大线程数
keepAliveTime	非核心线程空闲存活时间
unit	时间单位（TimeUnit枚举）
workQueue	任务队列（BlockingQueue实现）
threadFactory	线程创建工厂
handler	拒绝策略处理器

2.3 工作流程

提交任务时先创建核心线程
核心线程满后任务进入队列
队列满后创建非核心线程
达到最大线程数后触发拒绝策略

三、线程池创建方式

3.1 通过Executors工厂类

// 固定大小线程池
ExecutorService fixedPool = Executors.newFixedThreadPool(5);

// 单线程线程池
ExecutorService singleThreadPool = Executors.newSingleThreadExecutor();

// 可缓存线程池
ExecutorService cachedPool = Executors.newCachedThreadPool();

// 定时任务线程池
ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(3);

3.2 直接创建ThreadPoolExecutor

ThreadPoolExecutor customPool = new ThreadPoolExecutor(
    2, // corePoolSize
    5, // maximumPoolSize
    60, // keepAliveTime
    TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(10),
    Executors.defaultThreadFactory(),
    new ThreadPoolExecutor.AbortPolicy()
);

四、线程池使用实例

4.1 基本使用示例

public class BasicThreadPoolDemo {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        
        for (int i = 1; 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();
    }
}

4.2 带返回值的任务

public class CallableDemo {
    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        
        List<Future<Integer>> futures = new ArrayList<>();
        for (int i = 1; i <= 5; i++) {
            final int num = i;
            Future<Integer> future = executor.submit(() -> {
                System.out.println("计算数字: " + num);
                return num * num;
            });
            futures.add(future);
        }
        
        for (Future<Integer> future : futures) {
            System.out.println("结果: " + future.get());
        }
        
        executor.shutdown();
    }
}

4.3 定时任务示例

public class ScheduledTaskDemo {
    public static void main(String[] args) {
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
        
        // 延迟执行
        scheduler.schedule(() -> {
            System.out.println("延迟3秒执行");
        }, 3, TimeUnit.SECONDS);
        
        // 周期性执行
        scheduler.scheduleAtFixedRate(() -> {
            System.out.println("每隔2秒执行一次");
        }, 1, 2, TimeUnit.SECONDS);
        
        // 保持程序运行
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        scheduler.shutdown();
    }
}

五、线程池配置策略

5.1 线程数量配置

CPU密集型：核心数 + 1
IO密集型：核心数 * (1 + 平均等待时间/平均计算时间)

5.2 队列选择策略

队列类型	特点	适用场景
ArrayBlockingQueue	有界队列	控制资源消耗
LinkedBlockingQueue	无界队列	吞吐量优先
SynchronousQueue	直接传递	高响应要求
PriorityBlockingQueue	优先级队列	任务有优先级

5.3 拒绝策略

策略	行为
AbortPolicy	抛出RejectedExecutionException
CallerRunsPolicy	由调用线程执行任务
DiscardPolicy	直接丢弃任务
DiscardOldestPolicy	丢弃队列最前面的任务

六、线程池监控与管理

6.1 监控指标

ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(5);

// 获取活动线程数
int activeCount = executor.getActiveCount();

// 获取已完成任务数
long completedTaskCount = executor.getCompletedTaskCount();

// 获取队列中的任务数
int queueSize = executor.getQueue().size();

// 获取池中当前线程数
int poolSize = executor.getPoolSize();

6.2 自定义监控线程池

public class MonitorThreadPoolExecutor extends ThreadPoolExecutor {
    public MonitorThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
            long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }
    
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        System.out.printf("准备执行: %s 活动线程数: %d 队列大小: %d%n",
                t.getName(), getActiveCount(), getQueue().size());
    }
    
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        System.out.printf("执行完成: %s 活动线程数: %d 已完成任务: %d%n",
                Thread.currentThread().getName(), 
                getActiveCount(), getCompletedTaskCount());
    }
}

七、生产环境最佳实践

7.1 线程池命名

public class NamedThreadFactory implements ThreadFactory {
    private final String namePrefix;
    private final AtomicInteger threadNumber = new AtomicInteger(1);

    public NamedThreadFactory(String poolName) {
        namePrefix = poolName + "-thread-";
    }

    @Override
    public Thread newThread(Runnable r) {
        return new Thread(r, namePrefix + threadNumber.getAndIncrement());
    }
}

// 使用方式
ExecutorService namedPool = new ThreadPoolExecutor(
    4, 8, 60, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(),
    new NamedThreadFactory("order-process")
);

7.2 异常处理

executor.submit(() -> {
    try {
        // 业务代码
    } catch (Exception e) {
        // 记录日志
        logger.error("任务执行异常", e);
        // 可能需要重试或补偿
    }
});

7.3 优雅关闭

// 停止接收新任务
executor.shutdown();

try {
    // 等待现有任务完成
    if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
        // 强制关闭
        executor.shutdownNow();
    }
} catch (InterruptedException e) {
    executor.shutdownNow();
    Thread.currentThread().interrupt();
}

八、常见问题与解决方案

8.1 线程池大小不合理

问题表现： - CPU利用率低 - 任务堆积严重

解决方案： - 根据任务类型调整线程数 - 使用动态线程池（如Hippo、DynamicTp）

8.2 任务堆积导致OOM

问题表现： - 内存持续增长 - 最终OutOfMemoryError

解决方案： - 使用有界队列 - 设置合理的拒绝策略 - 监控队列大小

8.3 线程泄漏

问题表现： - 线程数持续增加不释放 - 最终无法创建新线程

解决方案： - 确保任务不会永久阻塞 - 设置合理的keepAliveTime - 使用ThreadPoolExecutor的allowCoreThreadTimeOut

九、高级应用场景

9.1 ForkJoinPool

public class FibonacciTask extends RecursiveTask<Long> {
    final long n;
    
    FibonacciTask(long n) { this.n = n; }
    
    @Override
    protected Long compute() {
        if (n <= 10) {
            return fibonacci(n);
        }
        FibonacciTask f1 = new FibonacciTask(n - 1);
        f1.fork();
        FibonacciTask f2 = new FibonacciTask(n - 2);
        return f2.compute() + f1.join();
    }
    
    private long fibonacci(long n) {
        if (n <= 1) return n;
        return fibonacci(n - 1) + fibonacci(n - 2);
    }
}

// 使用
ForkJoinPool pool = new ForkJoinPool();
FibonacciTask task = new FibonacciTask(30);
long result = pool.invoke(task);

9.2 CompletableFuture组合异步任务

CompletableFuture.supplyAsync(() -> queryFromDB(), dbPool)
    .thenApplyAsync(data -> processData(data), cpuPool)
    .thenAcceptAsync(result -> saveResult(result), ioPool)
    .exceptionally(ex -> {
        logger.error("处理失败", ex);
        return null;
    });

十、总结

Java线程池是并发编程的核心组件，合理使用线程池可以： 1. 显著提高系统性能 2. 优化资源利用率 3. 提供更好的可管理性

在实际开发中应该： - 根据业务场景选择合适的配置 - 实现完善的监控机制 - 遵循最佳实践避免常见陷阱

通过本文的各种实例，开发者可以全面掌握线程池的使用方法，构建高效稳定的并发系统。 “`

注：实际字数约为5600字（含代码），这里展示的是主要结构和核心内容。完整文章需要展开各部分说明和补充更多示例代码。