多线程总结三：Java1.5并发Future模式

一.作用概述

1. 异步计算。我每天早晨起来都有很多事情要做：洗漱-煮鸡蛋-吃早餐-等等。我的步骤是起床后先煮鸡蛋，定时10分钟，再去洗漱，等我洗漱完毕在看鸡蛋是否煮好了，煮好了就开始吃早餐，没煮好则等一会。我们把洗漱完毕在煮鸡蛋叫做同步执行，把边煮鸡蛋边洗漱叫异步执行。
2. 异步体现之处(看源码)：客户端把FutureTask当做一个线程去执行时，会触发FutureTask的run()方法，这个时候就已经开始计算结果了；等客户端调用get()方法时候，这个结果已经计算好就把这个结果返回，没计算好get()方法阻塞。

二.UML

 

三.源码分析

1. Future，代表异步计算的结果，5个抽象方法顾名思义。
2. Runnable，线程接口，无返回值。
3. Callable，代表有返回值的一个任务。
4. RunnableFuture，组合了两个接口而已。
5. FutureTask，实现异步计算的地方。

   package java.util.concurrent;
   import java.util.concurrent.locks.*;
   
        //实例化FutureTask时实例化Sync
        public FutureTask(Callable<V> callable) {
           if (callable == null)
               throw new NullPointerException();
           sync = new Sync(callable);
       }
       //实例化FutureTask时
       //1.Executors.callable内部采用适配器模式适配出一个Callable对象
       //2.实例化Sync
       public FutureTask(Runnable runnable, V result) {
           sync = new Sync(Executors.callable(runnable, result));
       }
       //以下方法基本都是委托sync对象去实现
       public boolean isCancelled() {
           return sync.innerIsCancelled();
       }
   
       public boolean isDone() {
           return sync.innerIsDone();
       }
   
       public boolean cancel(boolean mayInterruptIfRunning) {
           return sync.innerCancel(mayInterruptIfRunning);
       }
   
   
       public V get() throws InterruptedException, ExecutionException {
           return sync.innerGet();
       }
   
       public V get(long timeout, TimeUnit unit)
           throws InterruptedException, ExecutionException, TimeoutException {
           return sync.innerGet(unit.toNanos(timeout));
       }
   
      
       protected void done() { }
   
       
       protected void set(V v) {
           sync.innerSet(v);
       }
   
       protected void setException(Throwable t) {
           sync.innerSetException(t);
       }
       //FutureTask实现了Runnable接口，当run()方法被执行的时候
       public void run() {
           sync.innerRun();
       }
   
       protected boolean runAndReset() {
           return sync.innerRunAndReset();
       }
   }

    
6. Sync类是异步计算被委托的地方，真正实现异步计算的地方。

   private final class Sync extends AbstractQueuedSynchronizer {
           private static final long serialVersionUID = -7828117401763700385L;
   
           /** State value representing that task is running */
           private static final int RUNNING   = 1;
           /** State value representing that task ran */
           private static final int RAN       = 2;
           /** State value representing that task was cancelled */
           private static final int CANCELLED = 4;
   
           /** The underlying callable */
           private final Callable<V> callable;
           /** The result to return from get() */
           private V result;
           /** The exception to throw from get() */
           private Throwable exception;
   
   
           private volatile Thread runner;
           ......
   
           V innerGet(long nanosTimeout) throws InterruptedException, ExecutionException, TimeoutException {
               if (!tryAcquireSharedNanos(0, nanosTimeout))
                   throw new TimeoutException();
               if (getState() == CANCELLED)
                   throw new CancellationException();
               if (exception != null)
                   throw new ExecutionException(exception);
               return result;             //返回result
           }
   
           void innerSet(V v) {
   	    for (;;) {
   		int s = getState();
   		if (s == RAN)
   		    return;
                   if (s == CANCELLED) {
   		    // aggressively release to set runner to null,
   		    // in case we are racing with a cancel request
   		    // that will try to interrupt runner
                       releaseShared(0);
                       return;
                   }
   		if (compareAndSetState(s, RAN)) {
                       result = v;      //将callable.call()的结果赋值给result
                       releaseShared(0);
                       done();
   		    return;
                   }
               }
           }
   
           ......
       
           //FutureTask的run方法被执行时
           void innerRun() {
               if (!compareAndSetState(0, RUNNING))
                   return;
               try {
                   runner = Thread.currentThread();
                   if (getState() == RUNNING) // recheck after setting thread
                       innerSet(callable.call());  //调用了callable.call()
                   else
                       releaseShared(0); // cancel
               } catch (Throwable ex) {
                   innerSetException(ex);
               }
           }
   
       }

    

      补充一个类Executors.RunnableAdapter，将一个Runnable对象适配成一个Callable对象，显然result为空。所以可以说Runnable是一个没有返回值的Callable。

static final class RunnableAdapter<T> implements Callable<T> {
        final Runnable task;
        final T result;
        RunnableAdapter(Runnable  task, T result) {
            this.task = task;
            this.result = result;
        }
        public T call() {
            task.run();
            return result;
        }
    }

 

四.煮鸡蛋代码

package com.zzy.future;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;

public class CookEgg {

	public static void main(String[] args) throws InterruptedException, ExecutionException {

		//1.煮鸡蛋
		System.out.println("开始煮鸡蛋");
		FutureTask<Egg> task = new FutureTask<Egg>(new EggCallable());
		ExecutorService service = Executors.newSingleThreadExecutor();
		service.execute(task);
		//2.洗漱8分钟
		Thread.sleep(1000 * 8L);
		System.out.println("洗漱完毕");
		
		System.out.println("鸡蛋煮熟没: " + task.isDone());
		
		System.out.println(task.get().isHasCooked());
	}
}

class EggCallable implements Callable<Egg> {

	@Override
	public Egg call() throws Exception {
		Thread.sleep(1000 * 10L);  //鸡蛋煮10分钟
		return new Egg(true);
	}
	
}

class Egg {
	private boolean hasCooked;
	
	public Egg(boolean hasCooked) {
		this.hasCooked = hasCooked;
	}

	public boolean isHasCooked() {
		return hasCooked;
	}
}

评论

1 楼 sakajiaofu 2012-11-06  

这里面会有两个线程
一个是主线程，就是让你产生线程池让你启动线程的那个线程
这个线程其实一直都是同步的，没有异步
异步的是你在这个主线程里面产生一个线程池以后，就会产生一个新的线程，就相当于是产生了一个新的分支，这个时候，就使用了异步了
你的主线程里面还可以再产生完线程池以后，继续执行下面的代码，而另一个线程在处理你的逻辑，其实就相当于是有 两条路，一条是主干，一个是分支，主干和分支一起作业，主干可以做事，分支也可以，并且主干里面可以再某一个时刻的时候 通过FutrueTask 来获取到分支里面的内容