java ExecutorService线程池使用(ExecutorService/Completable异步+ExecutorService线程池)
文章目录
- 前置
- 线程池配置
- newFixedThreadPool
- newCachedThreadPool
- newScheduledThreadPool
- 自定义配置线程池
- 异步执行任务
- ExecutorService 提交无返回值任务
- ExecutorService 提交有返回值任务
- ScheduledExecutorService 周期性执行任务
- ScheduledExecutorService 提交延迟任务
- 线程池关闭
前置
线程池在多线程操作的场景中会被用到
使用 ExecutorService 时候,强烈建议自定义线程池!
一般开发中,CompletableFuture 可作为 Java 异步编程的首选工具,配合上 ExecutorService 线程池配置来用
线程池配置
newFixedThreadPool
使用简单配置,填写 5 核心线程数,同时最多也是 5 个线程,即 5 个线程始终保持连接不会被回收
该线程池会持有一个无边界的队列,存储的需要被执行的任务,比如某个时刻突然有大量的任务打进来,需要被执行,最开头打进来的 5 个任务会正在执行,其他的任务全都会进入队列中阻塞,此时就有可能造成 OOM,因为毕竟它使用的是无边界队列,在短时间内大量任务打进来的场景中是有可能造成 OOM 的
因此 newFixedThreadPool 更适合负载稳定的较简单的场景,或者无大量并发的简单场景。我更建议依据业务特性,自定义配置线程池!
ExecutorService pool = Executors.newFixedThreadPool(5);
newCachedThreadPool
使用简单配置,参数空,核心线程数 0,最大线程数 = Integer.MAX_VALUE,线程池会根据需要动态创建线程,使用 SynchronousQueue 无容量队列,一旦多个任务同时进来,一旦没有空闲的线程,就会创建新的线程
如果线程中任务执行时间较长,在短时间内有大量任务打进来,会造成瞬时创建非常多的线程同时执行,造成 CPU 飙升和内存 OOM,不适用流量激增的场景,但适合短时间内较大并发的任务
我更建议依据业务特性,自定义配置线程池!
ExecutorService pool = Executors.newCachedThreadPool(); // 动态调整线程数
newScheduledThreadPool
使用简单配置,填写 2 核心线程数,其后续可以执行周期性执行的线程操作
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
自定义配置线程池
使用自定义配置的线程池,
int corePoolSize = 5; // 核心线程数
int maxPoolSize = 10; // 最大线程数
long keepAliveTime = 60L; // 非核心线程空闲存活时间
TimeUnit unit = TimeUnit.SECONDS;
BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(100); // 有界队列
ThreadFactory threadFactory = Executors.defaultThreadFactory();
RejectedExecutionHandler handler = new ThreadPoolExecutor.AbortPolicy(); // 拒绝策略,默认就是拒绝策略// 创建自定义线程池
ExecutorService customPool = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler
);// 创建自定义线程池(周期性执行)
ScheduledExecutorService scheduler = new ScheduledThreadPoolExecutor(corePoolSize,maximumPoolSize,keepAliveTime,unit,workQueue,threadFactory,handler
);
- CPU 密集型:CPU 核心数 + 1(减少上下文切换)
- IO 密集型:2 * CPU 核心数(线程常阻塞,需更多线程)
- 根据系统资源设置,避免资源耗尽(如 corePoolSize * 1.5)
- 短任务系统:30-60 秒(快速回收)
- 长任务系统:5-10 分钟(减少创建开销)
- workQueue:ArrayBlockingQueue(有界队列,控制任务堆积)
- workQueue:SynchronousQueue(零容量,直接传递,适合高吞吐)
- workQueue:LinkedBlockingQueue(无界队列,可能导致 OOM)
- handler:
- AbortPolicy:严格不允许丢任务(默认)
- CallerRunsPolicy:由提交任务的线程执行,限流场景(减缓任务提交速度)
- DiscardPolicy:静默丢弃新任务,允许丢部分非关键任务(如日志采集)
- DiscardOldestPolicy:丢弃队列中最旧的任务并重试提交,实时性要求高的场景(如行情推送)
异步执行任务
ExecutorService 提交无返回值任务
customPool.execute(() -> {System.out.println("无返回值任务执行,线程: " + Thread.currentThread().getName());try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}
});
ExecutorService 提交有返回值任务
Future<String> future = customPool.submit(() -> {System.out.println("有返回值任务执行,线程: " + Thread.currentThread().getName());Thread.sleep(1000);return "任务结果";
});String result = future.get(); // 这个操作会阻塞主线程,表示一直等待到 customPool 的线程 submit 取得结果后主线程才会执行
ScheduledExecutorService 周期性执行任务
0:表示 初始延迟时间(即首次执行任务前的等待时间),如果设置为 0,任务会 立即执行(不需要等待)
// 提交一个每秒执行一次的周期性任务
ScheduledExecutorService scheduler = ...;
scheduler.scheduleAtFixedRate(() -> {System.out.println("周期性任务执行,当前线程: " + Thread.currentThread().getName());
}, 0, 1, TimeUnit.SECONDS);
ScheduledExecutorService 提交延迟任务
只会执行一次任务。它的作用是延迟执行一次任务
// 提交一个延迟5秒执行的一次性任务
ScheduledExecutorService scheduler = ...;
scheduler.schedule(() -> {System.out.println("延迟任务执行,当前线程: " + Thread.currentThread().getName());
}, 5, TimeUnit.SECONDS);
线程池关闭
pool.shutdown(); // 不再接受新任务,等待已提交任务完成
try {if (!pool.awaitTermination(60, TimeUnit.SECONDS)) { // 等待 60dpool.shutdownNow(); // 强制终止未完成的任务}
} catch (InterruptedException e) {pool.shutdownNow(); // 立即关闭!
}