任务的批量提交invokeAll两种方法的区别

简介: 任务的批量提交invokeAll两种方法的区别

ExecutorService的invokeAll方法有两种用法:


1.exec.invokeAll(tasks)


2.exec.invokeAll(tasks, timeout, unit)


其中tasks是任务集合,timeout是超时时间,unit是时间单位


两者都会堵塞,必须等待所有的任务执行完成后统一返回,一方面内存持有的时间长;另一方面响应性也有一定的影响,毕竟大家都喜欢看看刷刷的执行结果输出,而不是苦苦的等待;


但是方法二增加了超时时间控制,这里的超时时间是针对的所有tasks,而不是单个task的超时时间。如果超时,会取消没有执行完的所有任务,并抛出超时异常。相当于将每一个future的执行情况用一个list集合保存,当调用future.get()方法取值时和设置的timeout比较,是否超时。


InvokeAll方法处理一个任务的容器(collection),并返回一个Future的容器。两个容器具有相同的结构;


这里提交的任务容器列表和返回的Future列表存在顺序对应的关系。


invokeAll将Future添加到返回容器中,这样可以使用任务容器的迭代器,从而调用者可以将它表现的Callable与Future关联起来。


当所有任务都完成时、调用线程被中断时或者超过时限时,限时版本的invokeAll都会返回结果。超过时限后,任何尚未完成的任务都会被取消。


作为invokeAll的返回值,每个任务要么正常地完成,要么被取消。


invokeAll控制批量任务的时间期限的例子:

package com.thread;
import java.math.BigDecimal;
import java.sql.Time;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
/**
 * 批量任务的限时 invokeAll(tasks) 批量提交不限时任务
 * 
 * invokeAll(tasks, timeout, unit) 批量提交限时任务
 * 
 * InvokeAll方法处理一个任务的容器(collection),并返回一个Future的容器。两个容器具有相同的结构:
 * invokeAll将Future添加到返回的容器中,这样可以使用任务容器的迭代器,从而调用者可以将它表现的Callable与Future 关联起来。
 * 当所有任务都完成时、调用线程被中断时或者超过时限时,限时版本的invokeAll都会返回结果。 超过时限后,任务尚未完成的任务都会被取消。
 * 
 * @author hadoop
 *
 */
public class InvokeAllThread {
  // 固定大小的线程池,同时只能接受5个任务
  static ExecutorService mExecutor = Executors.newFixedThreadPool(5);
  /**
  * 计算价格的任务
  * @author hadoop
  *
  */
  private class QuoteTask implements Callable<BigDecimal> {
  public final double price;
  public final int num;
  public QuoteTask(double price, int num) {
    this.price = price;
    this.num = num;
  }
  @Override
  public BigDecimal call() throws Exception {
    Random r = new Random();
    long time = (r.nextInt(10) + 1) * 1000;
    Thread.sleep(time);
    BigDecimal d = BigDecimal.valueOf(price * num).setScale(2);
    System.out.println("耗时:" + time / 1000 + "s,单价是:" + price + ",人数是:"
      + num + ",总额是:" + d);
    return d;
  }
  }
  /**
  * 在预定时间内请求获得旅游报价信息
  * 
  * @return
  */
  public   void getRankedTravelQuotes() throws InterruptedException {
  List<QuoteTask> tasks = new ArrayList<QuoteTask>();
  // 模拟10个计算旅游报价的任务
  for (int i = 1; i <= 20; i++) {
    tasks.add(new QuoteTask(200, i) );
  }
  /**
   * 使用invokeAll方法批量提交限时任务任务 预期15s所有任务都执行完,没有执行完的任务会自动取消
   * 
   */
     List<Future<BigDecimal>> futures = mExecutor.invokeAll(tasks, 15, TimeUnit.SECONDS);
  // 报价合计集合
  List<BigDecimal> totalPriceList = new ArrayList<BigDecimal>();
  Iterator<QuoteTask> taskIter = tasks.iterator();
  for (Future<BigDecimal> future : futures) {
    QuoteTask task = taskIter.next();
    try {
    totalPriceList.add(future.get());
    } catch (ExecutionException e) {
    // 返回计算失败的原因
    // totalPriceList.add(task.getFailureQuote(e.getCause()));
    totalPriceList.add(BigDecimal.valueOf(-1));
     System.out.println("任务执行异常,单价是"+task.price+",人数是:"+task.num);
    } catch (CancellationException e) {
    // totalPriceList.add(task.getTimeoutQuote(e));
    totalPriceList.add(BigDecimal.ZERO);
     System.out.println("任务超时,取消计算,单价是"+task.price+",人数是:"+task.num);
    }
  }
  for (BigDecimal bigDecimal : totalPriceList) {
    System.out.println(bigDecimal);
  }
  mExecutor.shutdown();
  }
  public static void main(String[] args) {
  try {
    InvokeAllThread it = new InvokeAllThread();
    it.getRankedTravelQuotes();
  } catch (InterruptedException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
  }
  }
}


再来一个invokeAll不控制超时时间的小例子:

/**
 * 测试InvokeAll批量提交任务集
 * @throws InterruptedException 
 */
public  static void testInvokeAllThread() throws InterruptedException{
  ExecutorService exec = Executors.newFixedThreadPool(10);
  List<Callable<Integer>> tasks = new ArrayList<Callable<Integer>>();
  Callable<Integer> task = null;
  for (int i = 0; i < 20; i++) {
  task = new Callable<Integer>() {
    @Override
    public Integer call() throws Exception {
    int ran = new Random().nextInt(1000);
    Thread.sleep(ran);
    System.out.println(Thread.currentThread().getName()
      + " 休息了 " + ran);
    return ran;
    }
  };
  tasks.add(task);
  }
  long s = System.currentTimeMillis();
  List<Future<Integer>> results = exec.invokeAll(tasks);
  System.out.println("执行任务消耗了 :" + (System.currentTimeMillis() - s)
    + "毫秒");
  for (int i = 0; i < results.size(); i++) {
  try {
    System.out.println(results.get(i).get());
  } catch (Exception e) {
    e.printStackTrace();
  }
  }
  exec.shutdown();
}
目录
相关文章
|
缓存 Java 区块链
【Springboot】springboot项目替换掉默认的小叶子ico
【Springboot】springboot项目替换掉默认的小叶子ico
510 0
西门子S7-1200有什么功能特点?应用范围有哪些?CPU型号及模块类型有哪些?
S7-1200是西门子公司新推出的一款面向离散自动化系统和独立自动化系统的低端PLC。S7-1200采用了模块化设计,具备强大的工艺功能,适用于多种场合,可以满足不同的自动化需求。
西门子S7-1200有什么功能特点?应用范围有哪些?CPU型号及模块类型有哪些?
|
存储 SQL NoSQL
MongoDB 6.0 新特性概览
正如发布MongoDB 5.0时承诺的更快发布频率,年度大版本MongoDB 6.0也于2022年正式跟广大数据库爱好者们见面了。目前阿里云MongoDB已经完成了对6.0版本的适配工作,大家可以直接在官网控制台进行购买和尝鲜体验!
MongoDB 6.0 新特性概览
|
11月前
|
存储 JSON fastjson
再也不用心惊胆战地使用FastJSON了——序列化篇
本篇将主要介绍json序列化的详细流程。本文阅读的FastJSON源码版本为2.0.31。
3530 49
|
数据安全/隐私保护
Zookeeper快速入门(Zookeeper概述、安装、集群安装、选举机制、命令行操作、节点类型、监听器原理)(二)
Zookeeper快速入门(Zookeeper概述、安装、集群安装、选举机制、命令行操作、节点类型、监听器原理)(二)
|
12月前
|
数据可视化 数据挖掘 数据处理
实时计算Flink评测
本文介绍了Flink在实时计算领域的应用实践及核心功能评估,涵盖用户行为分析、电商羊毛党识别、实时销售数据分析三大场景,展示了Flink在处理实时数据流时的高效性、准确性和可靠性。同时,文章还深入探讨了Flink的统一数据处理、事件驱动处理、高容错性、高性能低延迟、灵活窗口操作及丰富API等核心功能,并指出了其优势与待改进之处,为用户提供全面的参考。
|
存储 SQL 关系型数据库
MySQL8 中文参考(二十)(1)
MySQL8 中文参考(二十)
235 3
|
SQL 关系型数据库 MySQL
MySQL 数据库中 CAST 函数如何使用?
MySQL 数据库中 CAST 函数如何使用?
795 0
|
Web App开发 Linux C++
Playwright系列(7):用VSCode 开始写Playwright 脚本
Playwright系列(7):用VSCode 开始写Playwright 脚本
2085 0
Playwright系列(7):用VSCode 开始写Playwright 脚本
|
SpringCloudAlibaba 监控 Java
SpringCloud Alibaba Sentinel实现熔断与限流--学习笔记
SpringCloud Alibaba Sentinel实现熔断与限流--学习笔记
244 0