当前位置： 首页 > news >正文

[Java线程池]ExecutorService|CompletionService的区别与选择

news 2025/8/24 15:11:47

这段时间对业务系统做了个性能测试，其中使用了较多线程池的技术，故此做一个技术总结。

这次总结的内容比较多，主要是四个：

  1. ExecutorService
  2. CompletionService
  3. Runnable
  4. Callable

前两个是线程池相关接口，后两个是多线程相关接口。在最后，我会说明什么情况下使用哪个接口，这两类接口如何搭配使用。

Tips：个人拙见，如有不对，请多多指正。

一、ExecutorService

ExecutorService是一个接口，继承自Executor。ExecutorService提供了一些常用操作和方法，但是ExecutorService是一个接口，无法实例化。
不过，Java提供了一个帮助类Executors，可以快速获取一个ExecutorService对象，并使用ExecutorService接口的一些方法。
ExecutorService
Executors帮助类提供了多个构造线程池的方法，常用的分为两类：

  1. 直接执行的
    • newCachedThreadPool
    • newFixedThreadPool
    • newSingleThreadExecutor
  2. 延迟或定时执行的
    • newScheduledThreadPool
    • newSingleThreadScheduledExecutor

Executors为每种方法提供了一个线程工厂重载。

（一）newCachedThreadPool

创建一个默认的线程池对象，里面的线程和重用，且在第一次使用的时候才创建。可以理解为线程优先模式，来一个创一个线程，直到线程处理完成后，再处理其他的任务。
Code：

package com.macro.boot.javaBuiltThreadPool;import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;public class MyExecutorService {public static void main(String[] args) {// 1. 使用帮助类
//        ExecutorService executorService = Executors.newCachedThreadPool();// 2. 提交任务
/*        for (int i = 0; i < 20; i++) {executorService.submit(new MyRunnable(i));}*/// 3. 重载方法测试test();}private static void test() {// 1. 使用帮助类ExecutorService executorService = Executors.newCachedThreadPool(new ThreadFactory() {int n = 1;@Overridepublic Thread newThread(Runnable r) {return new Thread(r, "线程正在执行 --->" + n++);}});// 2. 提交任务for (int i = 0; i < 20; i++) {executorService.submit(new MyRunnable(i));}}
}/*** 1. 线程类*/
class MyRunnable implements Runnable {private int id;public MyRunnable(int id) {this.id = id;}@Overridepublic void run() {String name = Thread.currentThread().getName();System.out.println(name + "正在执行..." + "--->" + id);}
}

输出：几乎是一下子就执行了，newCachedThreadPool会创建和任务数同等匹配的线程，直到处理完成任务的线程可以处理新增的任务。

（二）newFixedThreadPool

Code：创建一个可重用固定线程数量的线程池

package com.macro.boot.javaBuiltThreadPool;import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;/*** 创建一个可固定重用次数的线程池*/
public class MyNewFixedThreadPool {public static void main(String[] args) {
/*        // nThreads：线程数量ExecutorService es = Executors.newFixedThreadPool(5);for (int i = 0; i < 10; i++) {es.submit(new MyRunnable(i));}*/test();}private static void test() {ExecutorService es = Executors.newFixedThreadPool(5, new ThreadFactory() {int n = 1;@Overridepublic Thread newThread(Runnable r) {return new Thread(r, "线程" + n++);}});// 提交任务for (int i = 0; i < 10; i++) {es.submit(new MyRunnable(i));}}
}

（三）newSingleThreadExecutor

只有一个线程（线程安全）

package com.macro.boot.javaBuiltThreadPool;import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;public class MyNewSingleThreadExecutor {public static void main(String[] args) throws InterruptedException {
/*        ExecutorService es = Executors.newSingleThreadExecutor();for (int i = 0; i < 10; i++) {es.submit(new MyRunnable(i));}*/test();}private static void test() throws InterruptedException {ExecutorService es = Executors.newSingleThreadExecutor(new ThreadFactory() {int n = 1;@Overridepublic Thread newThread(Runnable r) {return new Thread(r, "线程" + n++);}});for (int i = 0; i < 10; i++) {Thread.sleep(100);es.submit(new MyRunnable(i));}}
}

（四）newScheduledThreadPool

怎么理解这个线程池的延迟时间？很简单，第一次执行的开始时间，加上延迟的时间，就是第二次执行的时间。

package com.macro.boot.ScheduledExecutorService;import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;public class MyScheduledExecutor {public static void main(String[] args) {ScheduledExecutorService sec = Executors.newScheduledThreadPool(4);for (int i = 0; i < 10; i++) {sec.schedule(new MyRunnable(i), 1, TimeUnit.SECONDS);}System.out.println("开始执行。。。");sec.shutdown();}
}class MyRunnable implements Runnable {private int id;@Overridepublic String toString() {return "MyRunnable{" +"id=" + id +'}';}public MyRunnable(int id) {this.id = id;}@Overridepublic void run() {String name = Thread.currentThread().getName();System.out.println(name + "执行了任务" + id);}
}

（五）newSingleThreadScheduledExecutor

newSingleThreadScheduledExecutor和newScheduledThreadPool的区别是，newSingleThreadScheduledExecutor的第二次执行时间，等于第一次开始执行的时间，加上执行线程所耗费的时间，再加上延迟时间，即等于第二次执行的时间。

二、CompletionService

CompletionService是一个接口。
当我们使用ExecutorService启动多个Callable时，每个Callable返回一个Future，而当我们执行Future的get方法获取结果时，会阻塞线程直到获取结果。
而CompletionService正是为了解决这个问题，它是Java8的新增接口，它的实现类是ExecutorCompletionService。CompletionService会根据线程池中Task的执行结果按执行完成的先后顺序排序，任务先完成的可优先获取到。
Code：

package com.macro.boot.completions;import java.util.concurrent.*;public class CompletionBoot {public static void main(String[] args) throws InterruptedException, ExecutionException {// 实例化线程池ExecutorService es = Executors.newCachedThreadPool();ExecutorCompletionService<Integer> ecs = new ExecutorCompletionService<>(es);for (int i = 0, j = 3; i < 20; i++) {ecs.submit(new CallableExample(i, j));}for (int i = 0; i < 20; i++) {// take：阻塞方法，从结果队列中获取并移除一个已经执行完成的任务的结果，如果没有就会阻塞，直到有任务完成返回结果。Integer integer = ecs.take().get();// 从结果队列中获取并移除一个已经执行完成的任务的结果，如果没有就会返回null，该方法不会阻塞。// Integer integer = ecs.poll().get();System.out.println(integer);}// 不要忘记关闭线程池es.shutdown();}
}
class CallableExample implements Callable<Integer> {/*** 使用构造方法获取变量* */private int a;private int b;public CallableExample(int a, int b) {this.a = a;this.b = b;}@Overridepublic Integer call() throws Exception {return a + b;}@Overridepublic String toString() {return "CallableExample{" +"a=" + a +", b=" + b +'}';}
}

三、Runnable

Runnable和Callable两者都是接口，但是也有区别：

  1. 实现Callable接口的任务线程能返回执行结果；而实现Runnable接口的任务线程不能返回结果；（重点）
  2. Callable接口的call()方法允许抛出异常；而Runnable接口的run()方法的异常只能在内部消化，不能继续上抛；

Code：

class MyRunnable02 implements Runnable {private int i;public MyRunnable02(int i) {this.i = i;}@Overridepublic void run() {String name = Thread.currentThread().getName();System.out.println(name + "执行了... ---> " + i);}@Overridepublic String toString() {return "MyRunnable{" +"i=" + i +'}';}
}

四、Callable

Code：

class CallableExample implements Callable<Integer> {/*** 使用构造方法获取变量* */private int a;private int b;public CallableExample(int a, int b) {this.a = a;this.b = b;}@Overridepublic Integer call() throws Exception {return a + b;}@Overridepublic String toString() {return "CallableExample{" +"a=" + a +", b=" + b +'}';}
}

五、Example

本次Demo：使用线程池，循环查询数据库500次。
在最开始的时候，是使用ExecutorServer + Future.get（因为查询数据库肯定需要获取结果，所以必须要用Callable，并且get到结果集）。但是get的阻塞操作，实在是太影响速度了，虽然考虑了两种手段去解决，但是都不了了之。
Code：（只贴线程池的代码，线程类和获取连接的类就不放了）

private void executorServerStart() throws SQLException, ClassNotFoundException, ExecutionException, InterruptedException {// get conTDConUtils tdConUtils = new TDConUtils();Connection con = tdConUtils.getCon();Statement statement = con.createStatement();// SQLString sql = "select last_row(value_double) from db1.tb1;";// ThreadPoolExecutorService es = Executors.newCachedThreadPool();// for eachint count = 500;for (int i = 0; i < count; i++) {Future<ResultSet> submit = es.submit(new MyThread(i, con, sql));ResultSet resultSet = submit.get();// printwhile (resultSet.next()) {System.out.printf("输出：时间：%s,值：%f \n", resultSet.getTimestamp(1), resultSet.getDouble(2));}}es.shutdown();// close resourcestdConUtils.close(con, statement);}

运行时间：8000ms +
改CompletionService：
Code：

private void completionServerStart() throws SQLException, ClassNotFoundException, InterruptedException, ExecutionException {// get conTDConUtils tdConUtils = new TDConUtils();Connection con = tdConUtils.getCon();Statement statement = con.createStatement();// SQLString sql = "select last_row(value_double) from db1.tb1;";// ThreadPoolExecutorService es = Executors.newCachedThreadPool();//构建ExecutorCompletionService,与线程池关联ExecutorCompletionService<ResultSet> ecs = new ExecutorCompletionService<ResultSet>(es);// for eachint count = 500;for (int i = 0; i < count; i++) {ecs.submit(new MyThread(i, con, sql));}for (int i = 0; i < count; i++) {// 通过take获取Future结果,此方法会阻塞ResultSet resultSet = ecs.take().get();while (resultSet.next()) {System.out.printf("输出：时间：%s,值：%f \n", resultSet.getTimestamp(1), resultSet.getDouble(2));}}es.shutdown();tdConUtils.close(con, statement);}

运行时间：300+ms

六、使用小结

分情况。
如果需要获取结果：线程使用Callable；
如果需要异步获取结果：线程池使用CompletionService。
如果不需要获取结果：线程使用Runnable；
如果需要阻塞获取结果：线程池使用ExecutorService。

http://www.lryc.cn/news/333236.html

相关文章：

  • MySQL-SQL编写练习：基本的SELECT语句
  • C++经典面试题目（十九）
  • acwing算法提高之图论--SPFA找负环
  • I2C驱动实验：测试I2C驱动是否与设备匹配
  • 5560.树的直径
  • Decoupled Multimodal Distilling for Emotion Recognition 论文阅读
  • 【css】使用display:inline-block后，元素间存在4px的间隔
  • 代码执行漏洞
  • SQLServer2022安装
  • vue2 配置@指向src
  • 用友U9 存在PatchFile.asmx接口任意文件上传漏洞
  • 如何卸载干净 IDEA（图文讲解）
  • 自动化运维（十）Ansible 之进程管理模块
  • 【leetcode279】完全平方数,动态规划解法
  • Java 元素排序（数组、List 集合）
  • 使用vite创建一个react18项目
  • 子集(迭代)(leetcode 78)
  • 汽车疲劳测试试验平台技术要求（北重厂家）
  • Redis -- 缓存雪崩问题
  • 【ARM 嵌入式 C 文件操作系列 20 -- 文件删除函数 remove 详细介绍】
  • LeetCode刷题之31.下一个排列
  • 【RISC-V 指令集】RISC-V 向量V扩展指令集介绍(九)- 向量定点算术指令
  • 【Java网络编程】IP网络协议与TCP、UDP网络传输层协议
  • C# 分布式自增ID算法snowflake（雪花算法）
  • commonJS和esModule的应用
  • （十一）RabbitMQ及SpringAMQP
  • STM32 M3内核寄存器概念
  • SQL语句的编写
  • Lecture 1~3 About Filter
  • 配置vscode链接linux