ExecutorService.invokeAny()和ExecutorService.invokeAll()的使用剖析

2024-09-04 12:48

本文主要是介绍ExecutorService.invokeAny()和ExecutorService.invokeAll()的使用剖析,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

ExecutorService.invokeAny()和ExecutorService.invokeAll()的使用剖析

ExecutorService是JDK并发工具包提供的一个核心接口,相当于一个线程池,提供执行任务和管理生命周期的方法。 ExecutorService接口中的大部分API都是比较容易上手使用的,本文主要介绍下invokeAll和invokeAll方法的特性和使用。我们先提供几个任务类:一个耗时任务,一个异常任务,一个短时任务。他们会在接下来的测试代码中使用。

package tasks;import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;public class SleepSecondsCallable implements Callable<String>
{
  private String name;  private int seconds;  public SleepSecondsCallable(String name, int seconds)
  {
    this.name = name;
    this.seconds = seconds;
  }  public String call() throws Exception
  {
    System.out.println(name + ",begin to execute");    try
    {
      TimeUnit.SECONDS.sleep(seconds);
    } catch (InterruptedException e)
    {
      System.out.println(name + " was disturbed during sleeping.");
      e.printStackTrace();
      return name + "_SleepSecondsCallable_failed";
    }    System.out.println(name + ",success to execute");    return name + "_SleepSecondsCallable_succes";
  }}

这是一个通过睡眠来模拟的耗时任务,该任务是可中断/可终止的任务,能够响应中断请求。

package tasks;import java.util.concurrent.Callable;public class ExceptionCallable implements Callable<String>
{  private String name = null;  public ExceptionCallable()
  {  }  public ExceptionCallable(String name)
  {
    this.name = name;
  }  @Override
  public String call() throws Exception
  {
    System.out.println("begin to ExceptionCallable.");    System.out.println(name.length());    System.out.println("end to ExceptionCallable.");    return name;
  }}

这是一个可能会在执行过程中,抛出空指针异常的任务。

package tasks;import java.util.Random;
import java.util.concurrent.Callable;public class RandomTenCharsTask implements Callable<String>
{  @Override
  public String call() throws Exception
  {
    System.out.println("RandomTenCharsTask begin to execute...");    StringBuffer content = new StringBuffer();    String base = "abcdefghijklmnopqrstuvwxyz0123456789";    Random random = new Random();    for (int i = 0; i < 10; i++)
    {
      int number = random.nextInt(base.length());
      content.append(base.charAt(number));
    }    System.out.println("RandomTenCharsTask complete.result=" + content);
    return content.toString();
  }}

这是一个正常的短时的任务,产生10个随机字符组成的字符串。

1.测试invokeAny()

第一种情况,向线程池提交2个耗时任务SleepSecondsCallable

/*** 提交的任务集合,一旦有1个任务正常完成(没有抛出异常),会终止其他未完成的任务*/
public static void invokeAny1() throws Exception
{
  ExecutorService executorService = Executors.newFixedThreadPool(3);  List<Callable<String>> tasks = new ArrayList<Callable<String>>();  tasks.add(new SleepSecondsCallable("t1", 2));
  tasks.add(new SleepSecondsCallable("t2", 1));  String result = executorService.invokeAny(tasks);  System.out.println("result=" + result);  executorService.shutdown();
}
程序的执行结果是:返回t2线程的执行结果t2_SleepSecondsCallable_succes,同时t1抛出java.lang.InterruptedException: sleep interrupted。

也就说: 一旦有1个任务正常完成(执行过程中没有抛异常),线程池会终止其他未完成的任务 。

第二种情况,向线程池提交3个异常任务ExceptionCallable

/**
* 没有1个正常完成的任务,invokeAny()方法抛出ExecutionException,封装了任务中元素的异常
* 
*/
public static void invokeAny2() throws Exception
{
  ExecutorService executorService = Executors.newFixedThreadPool(3);  List<Callable<String>> tasks = new ArrayList<Callable<String>>();  tasks.add(new ExceptionCallable());
  tasks.add(new ExceptionCallable());
  tasks.add(new ExceptionCallable());  String result = executorService.invokeAny(tasks);  System.out.println("result=" + result);  executorService.shutdown();
}
程序执行结果是:调用invokeAny()报错 java.util.concurrent.ExecutionException: java.lang.NullPointerException。

也就是说:

如果提交的任务列表中,没有1个正常完成的任务,那么调用invokeAny会抛异常,究竟抛的是哪儿个任务的异常,无关紧要

第三种情况:先提交3个异常任务,再提交1个正常的耗时任务

/**
* 有异常的任务,有正常的任务,invokeAny()不会抛异常,返回最先正常完成的任务
*/
public static void invokeAny3() throws Exception
{
  ExecutorService executorService = Executors.newFixedThreadPool(3);  List<Callable<String>> tasks = new ArrayList<Callable<String>>();  tasks.add(new ExceptionCallable());
  tasks.add(new ExceptionCallable());
  tasks.add(new ExceptionCallable());
  tasks.add(new ExceptionCallable());  tasks.add(new SleepSecondsCallable("t1", 2));  String result = executorService.invokeAny(tasks);  System.out.println("result=" + result);
  executorService.shutdown();
}

程序执行结果是:不会抛出任何异常,打印出t2任务的返回结果。也就是说:

invokeAny()和任务的提交顺序无关,只是返回最早正常执行完成的任务

第四种情况,测试下使用限时版本的invokeAny(),主要功能与不限时版本的差别不大

/*** 还没有到超时之前,所以的任务都已经异常完成,抛出ExecutionException<br>* 如果超时前满,还没有没有完成的任务,抛TimeoutException*/
public static void invokeAnyTimeout() throws Exception
{
  ExecutorService executorService = Executors.newFixedThreadPool(3);  List<Callable<String>> tasks = new ArrayList<Callable<String>>();  tasks.add(new ExceptionCallable());
  tasks.add(new ExceptionCallable());
  tasks.add(new ExceptionCallable());
  tasks.add(new ExceptionCallable());  String result = executorService.invokeAny(tasks, 2, TimeUnit.SECONDS);  System.out.println("result=" + result);  executorService.shutdown();
}
程序执行结果是:抛出ExecutionException。这个其实很合理,也很好理解。如果在超时之前,所有任务已经都是异常终止,那就没有必要在等下去了;如果超时之后,仍然有正在运行或等待运行的任务,那么会抛出TimeoutException。

最后我们来看下,JDK源码中ExecutorService.invokeAny的方法签名和注释

/**
  * Executes the given tasks, returning the result
  * of one that has completed successfully (i.e., without throwing
  * an exception), if any do. Upon normal or exceptional return,
  * tasks that have not completed are cancelled.
  * The results of this method are undefined if the given
  * collection is modified while this operation is in progress.
  *
  * @param tasks the collection of tasks
  * @return the result returned by one of the tasks
  * @throws InterruptedException if interrupted while waiting
  * @throws NullPointerException if tasks or any of its elements
  *	    are <tt>null</tt>
  * @throws IllegalArgumentException if tasks is empty
  * @throws ExecutionException if no task successfully completes
  * @throws RejectedExecutionException if tasks cannot be scheduled
  *	    for execution
  */<T> T invokeAny(Collection<? extends Callable<T>> tasks)
     throws InterruptedException, ExecutionException;

与我们测试结果一致,invokeAny()返回最先正常完成(without throwing exception)的任务直接结果;一旦有任务正常完成或者调用出现异常,线程池都会终止正在运行或等待运行(tasks that have not completed are cancelled)的任务。

2.测试invokeAll()

这个方法相对来说比较好理解,就是执行任务列表中的所有任务,并返回与每个任务对应的Futue。也就是说,任务彼此之间不会相互影响,可以通过future跟踪每一个任务的执行情况,比如是否被取消,是正常完成,还是异常完成,这主要使用Future类提供的API。

public static void testInvokeAll() throws Exception
{
  ExecutorService executorService = Executors.newFixedThreadPool(5);  List<Callable<String>> tasks = new ArrayList<Callable<String>>();
  tasks.add(new SleepSecondsCallable("t1", 2));
  tasks.add(new SleepSecondsCallable("t2", 2));
  tasks.add(new RandomTenCharsTask());
  tasks.add(new ExceptionCallable());  // 调用该方法的线程会阻塞,直到tasks全部执行完成(正常完成/异常退出)
  List<Future<String>> results = executorService.invokeAll(tasks);  // 任务列表中所有任务执行完毕,才能执行该语句
  System.out.println("wait for the result." + results.size());  executorService.shutdown();  for (Future<String> f : results)
  {
    // isCanceled=false,isDone=true
    System.out.println("isCanceled=" + f.isCancelled() + ",isDone="
        + f.isDone());    // ExceptionCallable任务会报ExecutionException
    System.out.println("task result=" + f.get());
  }
}

程序的执行结果和一些结论,已经直接写在代码注释里面了。invokeAll是一个阻塞方法,会等待任务列表中的所有任务都执行完成。不管任务是正常完成,还是异常终止,Future.isDone()始终返回true。通过

Future.isCanceled()可以判断任务是否在执行的过程中被取消。通过Future.get()可以获取任务的返回结果,或者是任务在执行中抛出的异常。

第二种情况,测试限时版本的invokeAll(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit)

/*** 可以通过Future.isCanceled()判断任务是被取消,还是完成(正常/异常)<br>* Future.isDone()总是返回true,对于invokeAll()的调用者来说,没有啥用*/
public static void testInvokeAllTimeout() throws Exception
{
  ExecutorService executorService = Executors.newFixedThreadPool(5);  List<Callable<String>> tasks = new ArrayList<Callable<String>>();
  tasks.add(new SleepSecondsCallable("t1", 2));
  tasks.add(new SleepSecondsCallable("t2", 2));
  tasks.add(new SleepSecondsCallable("t3", 3));
  tasks.add(new RandomTenCharsTask());  List<Future<String>> results = executorService.invokeAll(tasks, 1,
      TimeUnit.SECONDS);  System.out.println("wait for the result." + results.size());  for (Future<String> f : results)
  {
    System.out.println("isCanceled=" + f.isCancelled() + ",isDone="
        + f.isDone());
  }  executorService.shutdown();}

执行结果是:

wait for the result.4

isCanceled=true,isDone=true

isCanceled=true,isDone=true

isCanceled=true,isDone=true

isCanceled=false,isDone=true

也就是说给定的超时期满,还没有完成的任务会被取消,即Future.isCancelled()返回true;在超时期之前,无论是正常完成还是异常终止的任务, Future.is

Cancelled()返回false。

第三种情况,测试在等待invokeAll执行完成之前,线程被中断。

/*** 如果线程在等待invokeAll()执行完成的时候,被中断,会抛出InterruptedException<br>* 此时线程池会终止没有完成的任务,这主要是为了减少资源的浪费.*/
public static void testInvokeAllWhenInterrupt() throws Exception
{
  final ExecutorService executorService = Executors.newFixedThreadPool(5);  // 调用invokeAll的线程
  Thread invokeAllThread = new Thread() {    @Override
    public void run()
    {
      List<Callable<String>> tasks = new ArrayList<Callable<String>>();
      tasks.add(new SleepSecondsCallable("t1", 2));
      tasks.add(new SleepSecondsCallable("t2", 2));
      tasks.add(new RandomTenCharsTask());      // 调用线程会阻塞,直到tasks全部执行完成(正常完成/异常退出)
      try
      {
        List<Future<String>> results = executorService
            .invokeAll(tasks);
        System.out.println("wait for the result." + results.size());
      } catch (InterruptedException e)
      {
        System.out
            .println("I was wait,but my thread was interrupted.");
        e.printStackTrace();
      }    }
  };  invokeAllThread.start();  Thread.sleep(200);  invokeAllThread.interrupt();  executorService.shutdown();}

invokeAllThread 线程调用了ExecutorService.invokeAll(),在等待任务执行完成的时候,

invokeAllThread被别的线程中断了。这个时候,

ExecutorService.invokeAll()会抛出java.lang.InterruptedException,任务t1和t2都被终止抛出java.lang.InterruptedException: sleep interrupted。

也就是说一旦 ExecutorService.invokeAll()方法产生了异常,线程池中还没有完成的任务会被取消执行。

这篇关于ExecutorService.invokeAny()和ExecutorService.invokeAll()的使用剖析的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1136073

相关文章

中文分词jieba库的使用与实景应用(一)

知识星球:https://articles.zsxq.com/id_fxvgc803qmr2.html 目录 一.定义: 精确模式(默认模式): 全模式: 搜索引擎模式: paddle 模式(基于深度学习的分词模式): 二 自定义词典 三.文本解析   调整词出现的频率 四. 关键词提取 A. 基于TF-IDF算法的关键词提取 B. 基于TextRank算法的关键词提取

使用SecondaryNameNode恢复NameNode的数据

1)需求: NameNode进程挂了并且存储的数据也丢失了,如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2)故障模拟 (1)kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 (2)删除NameNode存储的数据(/opt/module/hadoop-3.1.4/data/tmp/dfs/na

Hadoop数据压缩使用介绍

一、压缩原则 (1)运算密集型的Job,少用压缩 (2)IO密集型的Job,多用压缩 二、压缩算法比较 三、压缩位置选择 四、压缩参数配置 1)为了支持多种压缩/解压缩算法,Hadoop引入了编码/解码器 2)要在Hadoop中启用压缩,可以配置如下参数

Makefile简明使用教程

文章目录 规则makefile文件的基本语法:加在命令前的特殊符号:.PHONY伪目标: Makefilev1 直观写法v2 加上中间过程v3 伪目标v4 变量 make 选项-f-n-C Make 是一种流行的构建工具,常用于将源代码转换成可执行文件或者其他形式的输出文件(如库文件、文档等)。Make 可以自动化地执行编译、链接等一系列操作。 规则 makefile文件

使用opencv优化图片(画面变清晰)

文章目录 需求影响照片清晰度的因素 实现降噪测试代码 锐化空间锐化Unsharp Masking频率域锐化对比测试 对比度增强常用算法对比测试 需求 对图像进行优化,使其看起来更清晰,同时保持尺寸不变,通常涉及到图像处理技术如锐化、降噪、对比度增强等 影响照片清晰度的因素 影响照片清晰度的因素有很多,主要可以从以下几个方面来分析 1. 拍摄设备 相机传感器:相机传

pdfmake生成pdf的使用

实际项目中有时会有根据填写的表单数据或者其他格式的数据,将数据自动填充到pdf文件中根据固定模板生成pdf文件的需求 文章目录 利用pdfmake生成pdf文件1.下载安装pdfmake第三方包2.封装生成pdf文件的共用配置3.生成pdf文件的文件模板内容4.调用方法生成pdf 利用pdfmake生成pdf文件 1.下载安装pdfmake第三方包 npm i pdfma

零基础学习Redis(10) -- zset类型命令使用

zset是有序集合,内部除了存储元素外,还会存储一个score,存储在zset中的元素会按照score的大小升序排列,不同元素的score可以重复,score相同的元素会按照元素的字典序排列。 1. zset常用命令 1.1 zadd  zadd key [NX | XX] [GT | LT]   [CH] [INCR] score member [score member ...]

git使用的说明总结

Git使用说明 下载安装(下载地址) macOS: Git - Downloading macOS Windows: Git - Downloading Windows Linux/Unix: Git (git-scm.com) 创建新仓库 本地创建新仓库:创建新文件夹,进入文件夹目录,执行指令 git init ,用以创建新的git 克隆仓库 执行指令用以创建一个本地仓库的

【北交大信息所AI-Max2】使用方法

BJTU信息所集群AI_MAX2使用方法 使用的前提是预约到相应的算力卡,拥有登录权限的账号密码,一般为导师组共用一个。 有浏览器、ssh工具就可以。 1.新建集群Terminal 浏览器登陆10.126.62.75 (如果是1集群把75改成66) 交互式开发 执行器选Terminal 密码随便设一个(需记住) 工作空间:私有数据、全部文件 加速器选GeForce_RTX_2080_Ti

【Linux 从基础到进阶】Ansible自动化运维工具使用

Ansible自动化运维工具使用 Ansible 是一款开源的自动化运维工具,采用无代理架构(agentless),基于 SSH 连接进行管理,具有简单易用、灵活强大、可扩展性高等特点。它广泛用于服务器管理、应用部署、配置管理等任务。本文将介绍 Ansible 的安装、基本使用方法及一些实际运维场景中的应用,旨在帮助运维人员快速上手并熟练运用 Ansible。 1. Ansible的核心概念