脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → CompletionService批量异步执行

Java中的CompletionService批量异步执行详解

 更新时间：2023年12月22日 09:05:54   作者：Java面试365  
这篇文章主要介绍了Java中的CompletionService批量异步执行详解,我们知道线程池可以执行异步任务,同时可以通过返回值Future获取返回值,所以异步任务大多数采用ThreadPoolExecutor+Future,需要的朋友可以参考下

前景引入

我们知道线程池可以执行异步任务，同时可以通过返回值Future获取返回值，所以异步任务大多数采用ThreadPoolExecutor+Future，如果存在如下情况，需要从任务一二三中获取返回值后，保存到数据库中，用异步逻辑实现代码应该如下所示。

public static void main(String[] args) throws ExecutionException, InterruptedException {
    ExecutorService executorService = Executors.newFixedThreadPool(3);
    Future<Integer> f1 = executorService.submit(() -> {
        System.out.println("执行任务一");
        return 1;
    });
    Future<Integer> f2 = executorService.submit(() -> {
        System.out.println("执行任务二");
        return 2;
    });
    Future<Integer> f3 = executorService.submit(() -> {
        System.out.println("执行任务三");
        return 3;
    });
    Integer r1 = f1.get();
    executorService.execute(()->{
        // 省略保存r1操作
        System.out.println(r1);
    });
    Integer r2 = f2.get();
    executorService.execute(()->{
        // 省略保存r2操作
        System.out.println(r2);
    });
    Integer r3 = f3.get();
    executorService.execute(()->{
        // 省略保存r3操作
        System.out.println(r3);
    });
    executorService.shutdown();
}

这样写的代码一点毛病没有，逻辑都是正常的，但如果存在任务一查询了比较耗时的操作，由于f1.get是阻塞执行，那么就算任务二和任务三已经返回结果，任务二的返回值和任务三的返回值都是不能保存到数据库的，因为f1.get将主线程阻塞了。

批量异步实现

那可以如何处理呢？可以采用万能的阻塞队列，任务先执行完毕的先入队，这样可以保证其它线程入库的速度不受影响，提高效率。

public static void main(String[] args) throws ExecutionException, InterruptedException {
    ExecutorService executorService = Executors.newFixedThreadPool(3);
    ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue(3);
    Future<Integer> f1 = executorService.submit(() -> {
        System.out.println("执行任务一");
        Thread.sleep(5000);
        return 1;
    });
    Future<Integer> f2 = executorService.submit(() -> {
        System.out.println("执行任务二");
        return 2;
    });
    Future<Integer> f3 = executorService.submit(() -> {
        System.out.println("执行任务三");
        Thread.sleep(3000);
        return 3;
    });
    executorService.execute(()->{
        try {
            Integer r1 = f1.get();
            // 阻塞队列入队操作
            queue.put(r1);
            System.out.println(r1);
        } catch (Exception e) {
            e.printStackTrace();
        }
    });
    executorService.execute(()->{
        try {
            Integer r2 = f2.get();
            queue.put(r2);
            System.out.println(r2);
        } catch (Exception e) {
            e.printStackTrace();
        }
    });
    executorService.execute(()->{
        try {
            Integer r3 = f3.get();
            queue.put(r3);
            System.out.println(r3);
        } catch (Exception e) {
            e.printStackTrace();
        }
    });
    // 循环次数不要使用queue.size限制，因为不同时刻queue.size值是有可能不同的
    for (int i = 0; i <3; i++) {
        Integer integer = queue.take();
        // 省略保存integer操作
        executorService.execute(()->{
            System.out.println("保存入库=="+integer);
        });
    }
    executorService.shutdown();
}

产生结果如下

image-20220307193401671

同样的在生产中不建议使用，因为SDK为我们提供了工具类CompletionService，CompletionService内部就维护了一个阻塞队列，唯一与上述代码实现有所区别的是，阻塞队列入库的是Future对象，其余原理类似。

CompletionService

如何创建CompletionService

CompletionService同样是一个接口，其具体实现为ExecutorCompletionService，创建CompletionService对象有两种方式

public ExecutorCompletionService(Executor executor);
public ExecutorCompletionService(Executor executor,BlockingQueue<Future<V>> completionQueue)

CompletionService对象的创建都是需要指定线程池，如果在创建时没有传入阻塞对象，那么会采用默认的LinkedBlockingQueue无界阻塞队列，如果应用到生产可能会产生OOM的情况，这是需要注意的。

CompletionService初体验

CompletionService如何做到批量执行异步任务呢，将上述场景采用CompletionService实现下

public static void main(String[] args) throws InterruptedException, ExecutionException {
    ExecutorService executorService = Executors.newFixedThreadPool(3);
    CompletionService completionService = new ExecutorCompletionService(executorService);
    Future<Integer> f1 = completionService.submit(() -> {
        System.out.println("执行任务一");
        Thread.sleep(5000);
        return 1;
    });
    Future<Integer> f2 = completionService.submit(() -> {
        System.out.println("执行任务二");
        return 2;
    });
    Future<Integer> f3 = completionService.submit(() -> {
        System.out.println("执行任务三");
        Thread.sleep(3000);
        return 3;
    });
    for (int i = 0; i <3 ; i++) {
        Future take = completionService.take();
        Integer integer = (Integer) take.get();
        executorService.execute(()->{
            System.out.println("执行入库=="+integer);
        });
    }
    executorService.shutdown();
}

CompletionService接口说明

CompletionService的方法不多，使用起来比较简单，方法签名如下

public static void main(String[] args) throws InterruptedException, ExecutionException {
    ExecutorService executorService = Executors.newFixedThreadPool(3);
    CompletionService completionService = new ExecutorCompletionService(executorService);
    Future<Integer> f1 = completionService.submit(() -> {
        System.out.println("执行任务一");
        Thread.sleep(5000);
        return 1;
    });
    Future<Integer> f2 = completionService.submit(() -> {
        System.out.println("执行任务二");
        return 2;
    });
    Future<Integer> f3 = completionService.submit(() -> {
        System.out.println("执行任务三");
        Thread.sleep(3000);
        return 3;
    });
    for (int i = 0; i <3 ; i++) {
        Future take = completionService.take();
        Integer integer = (Integer) take.get();
        executorService.execute(()->{
            System.out.println("执行入库=="+integer);
        });
    }
    executorService.shutdown();
}

总结

CompletionService主要是去解决无效等待的问题，如果一个耗时较长的任务在执行，那么可以采用这种方式避免无效的等待

CompletionService还能让异步任务的执行结果有序化，先执行完就先进入阻塞队列。

到此这篇关于Java中的CompletionService批量异步执行详解的文章就介绍到这了,更多相关CompletionService批量异步执行内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

相关文章

  • Spring Boot缓存问题分析及解决方案

    Spring Boot缓存问题分析及解决方案

    SpringBoot提供缓存支持,提升应用性能,但可能出现缓存不一致、缓存穿透、缓存击穿等问题,分析了缓存基本概念、SpringBoot缓存支持、常见缓存问题及解决方案,感兴趣的朋友跟随小编一起看看吧
    2024-09-09
  • Java基础将Bean属性值放入Map中的实例

    Java基础将Bean属性值放入Map中的实例

    这篇文章主要介绍了Java基础将Bean属性值放入Map中的实例的相关资料,需要的朋友可以参考下
    2017-07-07
  • MybatisPlus字段自动填充&乐观锁的方法实现

    MybatisPlus字段自动填充&乐观锁的方法实现

    MyBatis-Plus提供了便捷的自动填充功能,可用于自动处理创建时间、更新时间等字段,本文来介绍一下MybatisPlus字段自动填充&乐观锁的方法实现,感兴趣的可以了解一下
    2025-08-08
  • Java后端调用微信支付和支付宝支付的详细步骤

    Java后端调用微信支付和支付宝支付的详细步骤

    这篇文章主要介绍了Java后端如何调用微信支付和支付宝支付,涵盖了基本概念、配置步骤、代码示例以及注意事项,文中通过代码介绍的非常详细,需要的朋友可以参考下
    2025-04-04
  • Spring Boot + KingbaseES 四种主流的数据库连接池

    Spring Boot + KingbaseES 四种主流的数据库连接池

    文章介绍了四种主流的数据库连接池（DBCP、C3P0、Druid、HikariCP）在SpringBoot项目中与KingbaseES数据库的配置和使用,以及在实际开发中遇到的监控缺失和参数调优问题,并给出了相应的解决方案,感兴趣的朋友跟随小编一起看看吧
    2026-02-02
  • Log4j日志记录框架配置及用法解析

    Log4j日志记录框架配置及用法解析

    这篇文章主要介绍了Log4j日志记录框架配置及用法解析,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
    2020-07-07
  • Java导出Excel文件的方法

    Java导出Excel文件的方法

    项目里很多接口都涉及导出Excel文件,所以本文给大家总结了项目中导出Excel文件的方法,并通过代码示例讲解的非常详细,具有一定的参考价值,需要的朋友可以参考下
    2025-02-02
  • mybatis-plus报错net.sf.jsqlparser.statement.select.SelectBody的问题解决

    mybatis-plus报错net.sf.jsqlparser.statement.select.SelectBody的

    本文主要介绍了mybatis-plus报错net.sf.jsqlparser.statement.select.SelectBody的问题解决,具有一定的参考价值,感兴趣的可以了解一下
    2024-08-08
  • 基于@RequestParam name和value属性的区别

    基于@RequestParam name和value属性的区别

    这篇文章主要介绍了@RequestParam name和value属性的区别，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2021-08-08
  • SpringSecurity中的UserDetails和UserDetailsService接口详解

    SpringSecurity中的UserDetails和UserDetailsService接口详解

    这篇文章主要介绍了SpringSecurity中的UserDetails和UserDetailsService接口详解,UserDetailsService 在 Spring Security 中主要承担查询系统内用户、验证密码、封装用户信息和角色权限,需要的朋友可以参考下
    2023-11-11

最新评论