线程的并发工具类之Callable、Future 和 FutureTask原理和使用（九）

痛定思痛。 2022-12-19 13:15 88阅读 0赞

今天分析 Callable、Future 和 FutureTask原理和使用

1、Runnable  是一个接口，在它里面只声明了一个  run() 方法，由于  run() 方法返 回值为 void  类型，所以在执行完任务之后无法返回任何结果。 Callable 位于  java.util.concurrent  包下，它也是一个接口，在它里面也只声明 了一个方法，只不过这个方法叫做 call() ，这是一个泛型接口， call() 函数返回的 类型就是传递进来的 V  类型。 Future 就是对于具体的  Runnable  或者  Callable  任务的执行结果进行取消、查 询是否完成、获取结果。必要时可以通过 get  方法获取执行结果，该方法会阻塞 直到任务返回结果。

![2020110723023474.png][]

因为  Future  只是一个接口，所以是无法直接用来创建对象使用的，因此就 有了下面的 FutureTask 。

![20201107230256218.png][]

![20201107230303268.png][]

2、FutureTask  类实现了  RunnableFuture  接口， RunnableFuture  继承了  Runnable 接口和 Future  接口，而  FutureTask  实现了  RunnableFuture  接口。所以它既可以

作为  Runnable  被线程执行，又可以作为  Future  得到  Callable  的返回值。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

因此我们通过一个线程运行  Callable ，但是  Thread  不支持构造方法中传递 Callable 的实例，所以我们需要通过  FutureTask  把一个  Callable  包装成  Runnable ， 然后再通过这个 FutureTask  拿到  Callable  运行后的返回值。 要 new  一个  FutureTask  的实例，有两种方法

![20201107230348951.png][]

参见代码如   下

import java.util.Random;
    import java.util.concurrent.Callable;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.FutureTask;
    
    
    /**
     *类说明：演示Future等的使用
     */
    public class UseFuture {
    	
    	
    	/*实现Callable接口，允许有返回值*/
    	private static class UseCallable implements Callable<Integer>{
    		private int sum;
    		@Override
    		public Integer call() throws Exception {
    			System.out.println("Callable子线程开始计算！");  
    //			Thread.sleep(1000);
    	        for(int i=0 ;i<5000;i++){
    	        	if(Thread.currentThread().isInterrupted()) {
    					System.out.println("Callable子线程计算任务中断！");
    					return null;
    				}
    	            sum=sum+i;
    				System.out.println("sum="+sum);
    	        }  
    	        System.out.println("Callable子线程计算结束！结果为: "+sum);  
    	        return sum; 
    		}
    	}
    	
    	public static void main(String[] args) 
    			throws InterruptedException, ExecutionException {
    
    		UseCallable useCallable = new UseCallable();
    		//包装
    		FutureTask<Integer> futureTask = new FutureTask<>(useCallable);
    		Random r = new Random();
    		new Thread(futureTask).start();
    
    		Thread.sleep(1);
    		if(r.nextInt(100)>50){
    			System.out.println("Get UseCallable result = "+futureTask.get());
    		}else{
    			System.out.println("Cancel................. ");
    			futureTask.cancel(true);
    		}
    
    	}
    
    }

执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

3、结合手写线程池使用 核心代码：

@Resource(name="taskExecutor")
    	protected ThreadPoolTaskExecutor threadPool;
    
    List<Future<String>> taskResults = new ArrayList<Future<String>>();
    
    taskResults.add(process(threadPool, jyFileExchangeMonitor)); //将任务放入线程池中 进行处理
    
    //jyFileExchangeMonitor 普通对象随便传
    	private Future<String> process(final ThreadPoolTaskExecutor executor , JyFileExchangeMonitor jyFileExchangeMonitor) {
            return executor.submit(new Callable<String>() {
                @Override
                public String call() throws Exception {
                    try {
                    	handle(jyFileExchangeMonitor);
                    } catch (Exception e) {
                    	return jyFileExchangeMonitor.getLocation() + "error";
                    }
                    return jyFileExchangeMonitor.getLocation() + "success";
                }
            });
        }
    
    
    获取结果的相关核心代码：
    while (true) {
    							Thread.sleep(5000);
    				            boolean isAllDone = true;
    		
    				            for (Future<String> taskResult : taskResults) {
    				                isAllDone &= ( taskResult.isDone() || taskResult.isCancelled() );
    				            }
    				            logger.error("都处理完了吗？-------->"+isAllDone+"当前线程数 -->"+threadPool.getActiveCount());
    				            if (isAllDone) {
    				                // 任务都执行完毕，跳出循环 cvf
    				            	logger.error("线程处理完成 -->"+System.currentTimeMillis());
    				                break;
    				            }
    				            if((System.currentTimeMillis() - t1)/1000/60.0 > 5){ //若执行了三分钟还没有执行完，则取消此次任务
    				            	for (Future<String> taskResult : taskResults) {
    				            		taskResult.cancel(true);
    					            }
    				            	break;
    				            }
    				        }
    						//返回信息 线程处理的结果
    				        for(Future<String> taskResult : taskResults){
    				            String ex = null;
    				            try {
    				            	boolean cancelled = taskResult.isCancelled();
    				            	
    				            	if(!cancelled){
    				                ex = taskResult.get();
    				            	}
    				            	} catch (Exception e) {
    				                ex = ExceptionUtils.getStackTrace(e);
    				            }
    				            if(StringUtils.isNotEmpty(ex)){
    				                logger.error("线程处理 -->"+ex);
    				            }
    				        }

分析使用到此结束。

[2020110723023474.png]: /images/20221120/97997d9925e04f8d894b67657c97ced3.png
[20201107230256218.png]: /images/20221120/db7984ae12db493cb712ff62f22ad488.png
[20201107230303268.png]: https://img-blog.csdnimg.cn/20201107230303268.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20201107230319700.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA==,size_16,color_FFFFFF,t_70
[20201107230348951.png]: https://img-blog.csdnimg.cn/20201107230348951.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20201108202905283.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA==,size_16,color_FFFFFF,t_70