备案控制台
探索云世界
开发者社区 开发与运维 文章 正文

Java多线程 CompletionService和ExecutorCompletionService

2022-10-11 116
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： Java多线程 CompletionService和ExecutorCompletionService

一、说明

Future的不足


  • 当通过 .get() 方法获取线程的返回值时，会导致阻塞


  • 也就是和当前这个Future关联的计算任务真正执行完成的时候才返回结果


  • 新任务必须等待已完成任务的结果才能继续进行处理，会浪费很多时间，最好是谁最先执行完成谁最先返回


CompletionService的引入


  • 解决阻塞的问题


  • 以异步的方式一边处理新的线程任务，一边处理已完成任务的结果，将执行任务与处理任务分开进行处理


二、理解

CompletionService


  • java.util.concurrent包下CompletionService<V>接口，但并不继承Executor接口，仅有一个实现类ExecutorCompletionService用于管理线程对象


  • 更加有效地处理Future的返回值，避免阻塞，使用.submit()方法执行任务，使用.take()取得已完成的任务，并按照完成这些任务的时间顺序处理它们的结果


public interface CompletionService<V> {
    Future<V> submit(Callable<V> task);
    Future<V> submit(Runnable task, V result);
    Future<V> take() throws InterruptedException;
    Future<V> poll();
    Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException;
}


  • submit()方法用来执行线程任务


  • take()方法从队列中获取完成任务的Future对象，谁最先执行完成谁最先返回，获取到的对象再调用.get()方法获取结果


  • poll()方法获取并删除代表下一个已完成任务的 Future，如果不存在，则返回null，此无阻塞的效果


  • poll(long timeout, TimeUnit unti) timeout表示等待的最长时间，unit表示时间单位，在指定时间内还没获取到结果，则返回null


ExecutorCompletionService


  • java.util.concurrent包下ExecutorCompletionService<V>类实现CompletionService<V>接口，方法与接口相同


  • 比ExecutorService可以更精确和简便地完成异步任务的执行


  • executor执行任务，completionQueue保存异步任务执行的结果


public class ExecutorCompletionService<V> implements CompletionService<V> {
    private final Executor executor;
    private final AbstractExecutorService aes;
    private final BlockingQueue<Future<V>> completionQueue;
    ……
    Future<V> submit(Callable<V> task) 
    Future<V> submit(Runnable task, V result) 
    Future<V> take() throws InterruptedException
    Future<V> poll() 
    Future<V> poll(long timeout, TimeUnit unit)
    ……
}


  • completionQueue初始化了一个LinkedBlockingQueue类型的先进先出阻塞队列


    public ExecutorCompletionService(Executor executor) {
        if (executor == null)
            throw new NullPointerException();
        this.executor = executor;
        this.aes = (executor instanceof AbstractExecutorService) ?
            (AbstractExecutorService) executor : null;
        this.completionQueue = new LinkedBlockingQueue<Future<V>>();
    }


  • submit()方法中QueueingFuture是ExecutorCompletionService中的内部类


    public Future<V> submit(Callable<V> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<V> f = newTaskFor(task);
        executor.execute(new QueueingFuture<V>(f, completionQueue));
        return f;
    }


  • QueueingFuture将RunnableFuture实例对象赋值给了task，内部的done()方法将task添加到已完成阻塞队列中，调用take()或poll()方法获取已完成的Future


    private static class QueueingFuture<V> extends FutureTask<Void> {
        QueueingFuture(RunnableFuture<V> task,
                       BlockingQueue<Future<V>> completionQueue) {
            super(task, null);
            this.task = task;
            this.completionQueue = completionQueue;
        }
        private final Future<V> task;
        private final BlockingQueue<Future<V>> completionQueue;
        protected void done() { completionQueue.add(task); }
    }


三、实现

1.使用Future


创建CompletionServiceDemo类，创建好的线程对象，使用Executors工厂类来创建ExecutorService的实例(即线程池)，通过ThreadPoolExecutor的.submit()方法提交到线程池去执行，线程执行后，返回值Future可被拿到


public class CompletionServiceDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1.创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        // 2.创建Callable子线程对象任务
        Callable callable_1 = new Callable() {
            @Override
            public String call() throws Exception {
                Thread.sleep(5000);
                return ("我是Callable子线程 " +Thread.currentThread().getName()+ " 产生的结果 " );
            }
        };
        Callable callable_2 = new Callable() {
            @Override
            public String call() throws Exception {
                Thread.sleep(3000);
                return ("我是Callable子线程 " +Thread.currentThread().getName()+ " 产生的结果 " );
            }
        };
        Callable callable_3 = new Callable() {
            @Override
            public String call() throws Exception {
                Thread.sleep(1000);
                return ("我是Callable子线程 " +Thread.currentThread().getName()+ " 产生的结果 " );
            }
        };
        // 3.使用Future提交三个任务到线程池
        Future future_1 = executorService.submit(callable_1);
        Future future_2 = executorService.submit(callable_2);
        Future future_3 = executorService.submit(callable_3);
        // 4.获取返回值
        System.out.println("开始获取结果 " + getStringDate());
        System.out.println(future_1.get() + "" + getStringDate());
        System.out.println(future_2.get() + "" + getStringDate());
        System.out.println(future_3.get() + "" + getStringDate());
        System.out.println("结束 " + getStringDate());
        // 5.关闭线程池
        executorService.shutdown();
    }
    // 获取时间函数
    public static String getStringDate() {
        Date currentTime = new Date();
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("HH:mm:ss.SSS");
        String date = simpleDateFormat.format(currentTime);
        return date;
    }
}


future_1.get()会等待执行时间阻塞5秒再获取到结果，而在这5秒内future_2和future_3的任务已完成，所以会立马得到结果



2.使用ExecutorCompletionService


创建一个ExecutorCompletionService放入线程池实现CompletionService接口，将创建好的线程对象通过CompletionService提交任务和获取结果


public class CompletionServiceDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1.创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        // 2.创建一个ExecutorCompletionService放入线程池实现CompletionService接口
        CompletionService completionService = new ExecutorCompletionService(executorService);
        // 3.创建Callable子线程对象任务
        Callable callable_1 = new Callable() {
            @Override
            public String call() throws Exception {
                Thread.sleep(5000);
                return ("我是Callable子线程 " +Thread.currentThread().getName()+ " 产生的结果 " );
            }
        };
        Callable callable_2 = new Callable() {
            @Override
            public String call() throws Exception {
                Thread.sleep(3000);
                return ("我是Callable子线程 " +Thread.currentThread().getName()+ " 产生的结果 " );
            }
        };
        Callable callable_3 = new Callable() {
            @Override
            public String call() throws Exception {
                Thread.sleep(1000);
                return ("我是Callable子线程 " +Thread.currentThread().getName()+ " 产生的结果 " );
            }
        };
        // 3.使用CompletionService提交三个任务到线程池
        completionService.submit(callable_1);
        completionService.submit(callable_2);
        completionService.submit(callable_3);
        // 4.获取返回值
        System.out.println("开始获取结果 " + getStringDate());
        System.out.println(completionService.take().get() + "" + getStringDate());
        System.out.println(completionService.take().get() + "" + getStringDate());
        System.out.println(completionService.take().get() + "" + getStringDate());
        System.out.println("结束 " + getStringDate());
        // 5.关闭线程池
        executorService.shutdown();
    }
    // 获取时间函数
    public static String getStringDate() {
        Date currentTime = new Date();
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("HH:mm:ss.SSS");
        String date = simpleDateFormat.format(currentTime);
        return date;
    }
}


提交顺序是1-2-3，按照完成这些任务的时间顺序处理它们的结果，返回顺序是3-2-1



3.take()方法


take()方法从队列中获取完成任务的Future对象，会阻塞，一直等待线程池中返回一个结果，谁最先执行完成谁最先返回，获取到的对象再调用.get()方法获取结果


如果调用take()方法的次数大于任务数，会因为等不到有任务返回结果而阻塞，只有三个任务，第四次take等不到结果而阻塞



4.poll()方法


poll()方法不会去等结果造成阻塞，没有结果则返回null，接着程序继续往下运行

直接用completionService.poll().get()会引发 NullPointerException



创建一个循环，连续调用poll()方法，每次隔1秒调用，没有结果则返回null



public class CompletionServiceDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1.创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        // 2.创建一个ExecutorCompletionService放入线程池实现CompletionService接口
        CompletionService completionService = new ExecutorCompletionService(executorService);
        // 3.创建Callable子线程对象任务
        Callable callable_1 = new Callable() {
            @Override
            public String call() throws Exception {
                Thread.sleep(5000);
                return ("我是Callable子线程 " +Thread.currentThread().getName()+ " 产生的结果 " );
            }
        };
        Callable callable_2 = new Callable() {
            @Override
            public String call() throws Exception {
                Thread.sleep(3000);
                return ("我是Callable子线程 " +Thread.currentThread().getName()+ " 产生的结果 " );
            }
        };
        Callable callable_3 = new Callable() {
            @Override
            public String call() throws Exception {
                Thread.sleep(1000);
                return ("我是Callable子线程 " +Thread.currentThread().getName()+ " 产生的结果 " );
            }
        };
        // 3.使用CompletionService提交三个任务到线程池
        completionService.submit(callable_1);
        completionService.submit(callable_2);
        completionService.submit(callable_3);
        // 4.获取返回值
        System.out.println("开始获取结果 " + getStringDate());
        // 5.创建一个循环，连续调用poll()方法，间隔1秒
        for (int i = 0; i < 8; i++) {
            Future future = completionService.poll();
            if (future!=null){
                System.out.println(future.get() + getStringDate());
            }else {
                System.out.println(future+" "+getStringDate());
            }
            Thread.sleep(1000);
        }  
        System.out.println("结束 " + getStringDate());
        // 6.关闭线程池
        executorService.shutdown();
    }
    // 获取时间函数
    public static String getStringDate() {
        Date currentTime = new Date();
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("HH:mm:ss.SSS");
        String date = simpleDateFormat.format(currentTime);
        return date;
    }
}


5.poll(long timeout, TimeUnit unit)方法


poll(long timeout, TimeUnit unit)方法设置了等待时间，等待超时还没有结果就返回null


不使用 Thread.sleep(1000)，将等待时间设置成0.5秒，由于只有8次循环，也就是4秒执行时间，而callable_1需要执行5秒，获取不到结果则返回null



public class CompletionServiceDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1.创建线程池
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        // 2.创建一个ExecutorCompletionService放入线程池实现CompletionService接口
        CompletionService completionService = new ExecutorCompletionService(executorService);
        // 3.创建Callable子线程对象任务
        Callable callable_1 = new Callable() {
            @Override
            public String call() throws Exception {
                Thread.sleep(5000);
                return ("我是Callable子线程 " +Thread.currentThread().getName()+ " 产生的结果 " );
            }
        };
        Callable callable_2 = new Callable() {
            @Override
            public String call() throws Exception {
                Thread.sleep(3000);
                return ("我是Callable子线程 " +Thread.currentThread().getName()+ " 产生的结果 " );
            }
        };
        Callable callable_3 = new Callable() {
            @Override
            public String call() throws Exception {
                Thread.sleep(1000);
                return ("我是Callable子线程 " +Thread.currentThread().getName()+ " 产生的结果 " );
            }
        };
        // 3.使用CompletionService提交三个任务到线程池
        completionService.submit(callable_1);
        completionService.submit(callable_2);
        completionService.submit(callable_3);
        // 4.获取返回值
        System.out.println("开始获取结果 " + getStringDate());
        // 5.创建一个循环，连续调用poll()方法，间隔1秒
        for (int i = 0; i < 8; i++) {
            Future future = completionService.poll(500, TimeUnit.MILLISECONDS);
            if (future!=null){
                System.out.println(future.get() + getStringDate());
            }else {
                System.out.println(future+" "+getStringDate());
            }
        }
        System.out.println("结束 " + getStringDate());
        // 6.关闭线程池
        executorService.shutdown();
    }
    // 获取时间函数
    public static String getStringDate() {
        Date currentTime = new Date();
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("HH:mm:ss.SSS");
        String date = simpleDateFormat.format(currentTime);
        return date;
    }
}
文章标签：
Java
关键词：
Java线程
Java多线程
Java completionservice
线程completionservice
Yeats_Liao
目录
相关文章
请看我回答~
|
6天前
|
Java 程序员 开发者
深入理解Java并发编程：线程同步与锁机制
【4月更文挑战第30天】 在多线程的世界中，确保数据的一致性和线程间的有效通信是至关重要的。本文将深入探讨Java并发编程中的核心概念——线程同步与锁机制。我们将从基本的synchronized关键字开始，逐步过渡到更复杂的ReentrantLock类，并探讨它们如何帮助我们在多线程环境中保持数据完整性和避免常见的并发问题。文章还将通过示例代码，展示这些同步工具在实际开发中的应用，帮助读者构建对Java并发编程深层次的理解。
请看我回答~
17 0
请看我回答~
|
6天前
|
Java
Java并发编程：深入理解线程池
【4月更文挑战第30天】本文将深入探讨Java并发编程中的一个重要主题——线程池。我们将从线程池的基本概念入手，了解其工作原理和优势，然后详细介绍如何使用Java的Executor框架创建和管理线程池。最后，我们将讨论一些高级主题，如自定义线程工厂和拒绝策略。通过本文的学习，你将能够更好地理解和使用Java的线程池，提高你的并发编程能力。
请看我回答~
18 0
游客qf4jmczx4xu2y
|
1天前
|
Java
Java中的多线程编程：基础知识与实践
【5月更文挑战第5天】在现代软件开发中，多线程编程是一个重要的概念，尤其是在Java这样的多平台、高性能的编程语言中。通过多线程，我们可以实现并行处理，提高程序的运行效率。本文将介绍Java中多线程编程的基础知识，包括线程的概念、创建和控制方法，以及一些常见的多线程问题和解决方案。
游客qf4jmczx4xu2y
11 2
Yeats_Liao
|
4天前
|
存储 缓存 前端开发
Java串口通信技术探究3：RXTX库线程 优化系统性能的SerialPortEventListener类
Java串口通信技术探究3：RXTX库线程 优化系统性能的SerialPortEventListener类
Yeats_Liao
18 3
龙大吉
|
4天前
|
Java
JAVA难点包括异常处理、多线程、泛型和反射，以及复杂的分布式系统知识
JAVA难点包括异常处理、多线程、泛型和反射，以及复杂的分布式系统知识。入坑JAVA因它的面向对象特性、平台无关性、强大的标准库和活跃的社区支持。
龙大吉
18 2
肥猪肥猪-17824
|
4天前
|
Java 调度 开发者
Java中的多线程编程：基础与实践
【5月更文挑战第2天】本文将深入探讨Java中的多线程编程，从基础概念到实际应用，为读者提供全面的理解和实践指导。我们将首先介绍线程的基本概念和重要性，然后详细解析Java中实现多线程的两种主要方式：继承Thread类和实现Runnable接口。接着，我们将探讨线程同步的问题，包括synchronized关键字和Lock接口的使用。最后，我们将通过一个实际的生产者-消费者模型来演示多线程编程的实践应用。
肥猪肥猪-17824
15 3
shuj
|
4天前
|
安全 Java 程序员
Java中的多线程编程：从理论到实践
【5月更文挑战第2天】 在计算机科学中，多线程编程是一项重要的技术，它允许多个任务在同一时间段内并发执行。在Java中，多线程编程是通过创建并管理线程来实现的。本文将深入探讨Java中的多线程编程，包括线程的概念、如何创建和管理线程、以及多线程编程的一些常见问题和解决方案。
shuj
15 1
aliyun8599273441-30642
|
5天前
|
存储 安全 Java
深入理解Java并发编程：线程安全与性能优化
【5月更文挑战第1天】本文将深入探讨Java并发编程的核心概念，包括线程安全和性能优化。我们将详细分析线程安全问题的根源，以及如何通过合理的设计和编码实践来避免常见的并发问题。同时，我们还将探讨如何在保证线程安全的前提下，提高程序的并发性能，包括使用高效的同步机制、减少锁的竞争以及利用现代硬件的并行能力等技术手段。
aliyun8599273441-30642
26 7
天下无贼001
|
5天前
|
并行计算 Java 数据处理
Java中的多线程编程：基础知识与实践
【5月更文挑战第1天】本文将深入探讨Java中的多线程编程，包括其基本概念、实现方式以及实际应用。我们将从理论和实践两个角度出发，详细解析线程的创建、启动、控制以及同步等关键问题，并通过实例代码演示如何在Java中有效地使用多线程。
天下无贼001
20 6
游客uul6wusao3sds
|
5天前
|
Java 程序员
Java中的多线程编程：从理论到实践
【5月更文挑战第1天】 在现代计算机科学中，多线程编程是一个重要的概念，它允许程序员在同一程序中并行运行多个任务。Java作为一种广泛使用的编程语言，提供了一套丰富的多线程编程工具。本文将介绍Java中多线程编程的基本概念，包括线程的创建、启动、控制和同步，以及一些常见的多线程问题和解决方案。
游客uul6wusao3sds
16 2

热门文章

最新文章

  • 1
    SpringBoot实现多线程定时任务动态定时任务配置文件配置定时任务
  • 2
    Java线程池ThreadPoolExcutor源码解读详解07-阻塞队列之LinkedTransferQueue
  • 3
    安卓多线程和并发处理：提高应用效率
  • 4
    Java线程池ThreadPoolExcutor源码解读详解03-阻塞队列之LinkedBlockingQueue
  • 5
    Python网络编程基础（Socket编程）多线程/多进程服务器编程
  • 6
    Java面试必考题之线程的生命周期，结合源码，透彻讲解!
  • 7
    Java并发编程：深入理解线程池的原理与实践
  • 8
    深入理解Java并发编程：线程池的应用与优化
  • 9
    Python多线程、多进程与协程面试题解析
  • 10
    Java中的多线程编程：基础知识与实践
  • 1
    [Java基础]——类的生命周期
    13
  • 2
    [java进阶]——泛型类、泛型方法、泛型接口、泛型的通配符
    25
  • 3
    【java进阶】Java中线程的实现方式
    21
  • 4
    【java进阶】集合的三种遍历(迭代器、增强for、Lambda)
    18
  • 5
    [java进阶]——方法引用改写Lambda表达式
    30
  • 6
    深入理解 Java 多线程、Lambda 表达式及线程安全最佳实践
    120
  • 7
    java base64转doc文件
    26
  • 8
    [JavaWeb]——过滤器filter与拦截器Interceptor的使用、执行过程、区别
    40
  • 9
    Java中的异常语法知识居然这么好玩！后悔没有早点学习
    34
  • 10
    Java必刷入门递归题×5（内附详细递归解析图）
    23

    • 相关课程

    更多
  • Java面试疑难点解析 - 面试技巧及语言基础
  • Java面试疑难点解析 - 系统架构及项目设计
  • Java编程入门
  • Java面向对象编程
  • Java高级编程

    • 相关电子书

    更多
  • Spring Cloud Alibaba - 重新定义 Java Cloud-Native
  • The Reactive Cloud Native Arch
  • 多IO线程优化版

    • 相关实验场景

    更多
  • 基于ECS搭建Java Web开发环境
  • 使用Aliyun Java Intializr进行项目开发
  • 搭建Java Web开发环境
  • 部署基于Dragonwell的Java运行环境
  • 阿里云平台上进行Java程序的编译与运行
    • 下一篇
    将Stable Diffusion模型文件转存到FC环境的NAS