java ExecutorService的invokeAll方法有两种用法 +价格超时计算

左手的ㄟ右手 2022-10-07 06:48 125阅读 0赞

exec.invokeAll(tasks)

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的返回值，每个任务要么正常地完成，要么被取消。

示例：  
控制批量任务的时间期限

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();  
        \}  
    \}  
   
\}  
  
执行结果：  
耗时：3s,单价是：200.0,人数是：5，总额是：1000.00  
耗时：3s,单价是：200.0,人数是：3，总额是：600.00  
耗时：3s,单价是：200.0,人数是：7，总额是：1400.00  
耗时：7s,单价是：200.0,人数是：4，总额是：800.00  
耗时：4s,单价是：200.0,人数是：6，总额是：1200.00  
耗时：1s,单价是：200.0,人数是：9，总额是：1800.00  
耗时：10s,单价是：200.0,人数是：1，总额是：200.00  
耗时：10s,单价是：200.0,人数是：2，总额是：400.00  
耗时：3s,单价是：200.0,人数是：11，总额是：2200.00  
耗时：4s,单价是：200.0,人数是：10，总额是：2000.00  
耗时：2s,单价是：200.0,人数是：12，总额是：2400.00  
耗时：6s,单价是：200.0,人数是：8，总额是：1600.00  
耗时：3s,单价是：200.0,人数是：13，总额是：2600.00  
耗时：2s,单价是：200.0,人数是：17，总额是：3400.00  
任务超时，取消计算,单价是200.0，人数是：14  
任务超时，取消计算,单价是200.0，人数是：15  
任务超时，取消计算,单价是200.0，人数是：16  
任务超时，取消计算,单价是200.0，人数是：18  
任务超时，取消计算,单价是200.0，人数是：19  
任务超时，取消计算,单价是200.0，人数是：20  
200.00  
400.00  
600.00  
800.00  
1000.00  
1200.00  
1400.00  
1600.00  
1800.00  
2000.00  
2200.00  
2400.00  
2600.00  
0  
0  
0  
3400.00  
0  
0  
0

Process finished with exit code 0  
  
不控制超时时间期限

/\*\*  
 \* 测试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();  
\}  
  
执行结果  
pool-2-thread-8 休息了 20  
pool-2-thread-3 休息了 68  
pool-2-thread-2 休息了 278  
pool-2-thread-6 休息了 282  
pool-2-thread-1 休息了 310  
pool-2-thread-10 休息了 343  
pool-2-thread-8 休息了 448  
pool-2-thread-10 休息了 155  
pool-2-thread-9 休息了 568  
pool-2-thread-5 休息了 609  
pool-2-thread-9 休息了 99  
pool-2-thread-10 休息了 175  
pool-2-thread-8 休息了 271  
pool-2-thread-7 休息了 881  
pool-2-thread-4 休息了 927  
pool-2-thread-3 休息了 886  
pool-2-thread-5 休息了 407  
pool-2-thread-1 休息了 796  
pool-2-thread-2 休息了 856  
pool-2-thread-6 休息了 948  
执行任务消耗了 ：1232毫秒  
310  
278  
68  
927  
609  
282  
881  
20  
568  
343  
448  
886  
856  
948  
796  
155  
271  
175  
99  
407

Process finished with exit code 0  
————————————————  
版权声明：本文为CSDN博主「梦和远方」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。  
原文链接：https://blog.csdn.net/weixin\_42614447/article/details/88682854