多线程之Callable/Future获取多个任务结果并进行汇总

许多不能立即获得计算结果的延迟计算，都可以使用Callable/Future这种形式。例如数据库查询等。Callable是对任务的一种抽象，通过在Executor中包含封装的Callable任务来执行，将结果封装为Future。Future提供了相应的方法判断是否已经完成，以及获取任务的结果(future.get())等方便进行操作。

Get方法的行为取决于任务的状态（尚未开始、正在运行、已完成）。如果任务已经完成，那么get会立即返回或者抛出一个Exception，如果任务没有完成，那么get将阻塞并直到任务完成。如果任务抛出了异常，那么get将该异常封装为ExecutionException并重新抛出。如果任务被取消，那么get将抛出CanclellationException。如果get抛出了ExecutionException，那么可以通过getCause来获得被封装的初始异常。

尚一章的例子，如果用这个方法写将方便许多，现尝试如下：

package com.cc.mutilineExample.callableAndFuture.test3_future_every;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * @ClassName: FutureResponseTest 
 * @Description: 根据交易请求的额度和方式，进行扣款汇总计算
 * 计算交易一共扣款的数额：
 * 		利用Callable和Future来计算，future.get()会获取任务结果，如果没有获取到会一直阻塞直到任务结束。
 *
 * 想优化一下，即每一个都一个future  这样会更快
 * 
 * 这样虽可行，但是却有些繁琐。还有一种更好的办法：CompletionService
 * 等理论学习后，继续写一个例子
 * @author CC  
 * @date 2018年12月6日 上午10:44:20 
 * @version V1.0 
 */
public class FutureResponseTest {
	//线程池
	private final static ExecutorService executor = Executors.newCachedThreadPool();

	//模拟第三方服务
	public static double requestForService(Request request) throws InterruptedException, Exception{
		if(null == request) {
			throw new Exception("请求为空！");
		}
		if(request.getParam() <= 0) {
			throw new Exception("参数小于0，无法进行扣款！" + request);
		}
		
		System.out.println("开始处理请求...");
		
		//为了简便直接返回一个结果即可
		double result = 0.0;
		if("WeiXin".equals(request.getMethod())) {
			System.out.println("微信支付扣3%");
//			result = request.getParam() * 0.03;//double类型计算结果不准确  例如17 * 0.05 返回  扣款数  0.8500000000000001
			result = new BigDecimal(String.valueOf(request.getParam())).multiply(new BigDecimal("0.03")).doubleValue();
		}else {
			System.out.println("其他支付直接扣5%");
			result = new BigDecimal(String.valueOf(request.getParam())).multiply(new BigDecimal("0.05")).doubleValue();
		}
		
		//模拟-使消耗时间长一些
		Thread.sleep(3000);
		System.out.println(request + " 返回扣款结果：" + result);
		return result;
	}
	
	
	//调度请求，获得返回结果，并进行汇总处理
	public static void main(String[] args) throws Exception {
		final String[] methodStr = new String[] {"WeiXin","ZhiFuBao","WangYin"};
		final String[] serviceStr = new String[] {"TaoBao","JingDong","TianMao"};
		
		//为了方便，我们将请求先初始化完毕
		final List<Request> requestList = new ArrayList<Request>();
		for (int i = 0; i < 20; i++) {
			Request request = new Request();
			request.setMethod(methodStr[(int) (Math.random() * 3)]);
			request.setParam((int) (Math.random() * 300));
			request.setServieName(serviceStr[(int) (Math.random() * 3)]);
			requestList.add(request);
		}
		
		long startTime = System.currentTimeMillis();//开始时间
		
		//累积计算所有请求的总扣款数--计算任务提前开始且每个都是分开的不相互影响
		List<Future<Double>> futureList = new ArrayList<Future<Double>>();
		for (int i = 0; i < requestList.size(); i++) {
			Request request = requestList.get(i);
			Callable<Double> task = new Callable<Double>() {
				@Override
				public Double call() throws Exception {
					return  requestForService(request);
				}
			};
			Future<Double> future = executor.submit(task);
			futureList.add(future);
		}

		try {
			
			BigDecimal sum = BigDecimal.ZERO;//同理  double计算结果不精确
			
			//方法get具有“状态依赖”的内在特性，因而调用者不需要知道任务的状态，此外在任务提交和获得
			//结果中包含的安全发布属性也确保了这个方法是线程安全的。
			//获得结果
			for (int i = 0; i < futureList.size(); i++) {
				//提交任务请求
				double payMent = futureList.get(i).get();
				sum = sum.add(new BigDecimal(String.valueOf(payMent)));
			}
			System.out.println("一共扣款了多少钱？" + sum.doubleValue());
		} catch (InterruptedException e) {
			// TODO: 任务调用get的线程在获得结果之前被中断
			//重新设置线程的中断状态
			Thread.currentThread().interrupt();
			System.out.println("任务调用get的线程在获得结果之前被中断！" + e);
		} catch (ExecutionException e) {
			throw launderThrowable(e.getCause());
		} finally {
			executor.shutdown();
		}
		
		
		long endTime = System.currentTimeMillis();//结束时间
		System.out.println("消耗时间：" + (endTime - startTime) + "毫秒！");
	}



	/**
	 * @Title: launderThrowable 
	 * @Description: 任务执行过程中遇到异常，根据包装的ExecutionException重新抛出异常，并打印异常信息
	 * @param @param cause
	 * @param @return   
	 * @return Exception  
	 * @throws
	 * @author CC
	 * @date 2018年12月7日 上午9:36:27
	 * @version V1.0
	 */
	private static Exception launderThrowable(Throwable cause) {
		//抛出
		Exception exception = new Exception("任务执行过程中遇到异常！" + cause);
		//打印
		cause.printStackTrace();
		return exception;
	}
}

运行结果：

=正常运行结果========

开始处理请求…

其他支付直接扣5%

开始处理请求…

微信支付扣3%

开始处理请求…

其他支付直接扣5%

开始处理请求…

微信支付扣3%

开始处理请求…

其他支付直接扣5%

开始处理请求…

微信支付扣3%

开始处理请求…

其他支付直接扣5%

开始处理请求…

其他支付直接扣5%

开始处理请求…

其他支付直接扣5%

开始处理请求…

微信支付扣3%

开始处理请求…

其他支付直接扣5%

开始处理请求…

微信支付扣3%

开始处理请求…

其他支付直接扣5%

开始处理请求…

其他支付直接扣5%

开始处理请求…

其他支付直接扣5%

开始处理请求…

其他支付直接扣5%

开始处理请求…

微信支付扣3%

开始处理请求…

微信支付扣3%

开始处理请求…

其他支付直接扣5%

Request [method=ZhiFuBao, servieName=TaoBao, param=40] 返回扣款结果：2.0

Request [method=ZhiFuBao, servieName=JingDong, param=82] 返回扣款结果：4.1

Request [method=WeiXin, servieName=TaoBao, param=243] 返回扣款结果：7.29

Request [method=ZhiFuBao, servieName=JingDong, param=29] 返回扣款结果：1.45

Request [method=WangYin, servieName=TianMao, param=97] 返回扣款结果：4.85

Request [method=WeiXin, servieName=TaoBao, param=200] 返回扣款结果：6.0

Request [method=ZhiFuBao, servieName=TaoBao, param=240] 返回扣款结果：12.0

Request [method=ZhiFuBao, servieName=TaoBao, param=40] 返回扣款结果：2.0

Request [method=WangYin, servieName=TaoBao, param=102] 返回扣款结果：5.1

Request [method=WeiXin, servieName=JingDong, param=140] 返回扣款结果：4.2

Request [method=ZhiFuBao, servieName=TaoBao, param=131] 返回扣款结果：6.55

Request [method=ZhiFuBao, servieName=JingDong, param=219] 返回扣款结果：10.95

Request [method=WeiXin, servieName=TianMao, param=233] 返回扣款结果：6.99

Request [method=WeiXin, servieName=JingDong, param=203] 返回扣款结果：6.09

Request [method=WeiXin, servieName=TaoBao, param=120] 返回扣款结果：3.6

Request [method=ZhiFuBao, servieName=TaoBao, param=139] 返回扣款结果：6.95

Request [method=WangYin, servieName=TaoBao, param=244] 返回扣款结果：12.2

Request [method=ZhiFuBao, servieName=TaoBao, param=236] 返回扣款结果：11.8

Request [method=WangYin, servieName=JingDong, param=130] 返回扣款结果：6.5

Request [method=WeiXin, servieName=TaoBao, param=205] 返回扣款结果：6.15

一共扣款了多少钱？126.77

消耗时间：3010毫秒！

可以发现，同样实现了效率提升。且代码方便了许多。

多线程中为了简便Callable/Future对计算完成后获得结果，这种程序更好的支持，提供了一种新的接口CompletionService。将BlockingQueue与FutureTask很好地结合起来，下一章例子介绍。

多线程之Callable/Future获取多个任务结果并进行汇总

继续阅读

Java线程的概念：什么是线程？

2023爬虫学习笔记 -- 多线程操作

java socket多线程聊天室(一)

This application is modifying the autolayout engine from a background thread, which can lead

Socket实现聊天室（二）

网络编程----TCP实现聊天室项目前言

OO设计的重要原则

【一】深入理解Java高并发 - 并发的设计原则

避免条件竞争的三种方法

线程死锁的触发、修复、避免策略一、什么是死锁二、死锁例子三、死锁的必要条件四、定位死锁五、死锁的修复六、避免死锁的策略七、活锁八、总结

Go 基础之 Goroutine

ThreadLocal线程局部变量-多线程与高并发

python两种方法解决线程冲突问题线程冲突起因解决方案

C++ 多线程用条件变量确定线程的执行顺序而不是使用 sleep(1)

C#多线程——前台线程和后台线程

线程同步，可重入锁，synchronized