实验九：java Future 接口

下面讲一下java 的future 接口

 

我们都知道java 实现多线程的机制， 1.5之前调用runnable 接口。

 

1.5之后我们可以同过别的方式实现多线程。

 

如果我们想执行另外一个线程 比如说调用class A 的B()方法，

 

最好的方式就是调用A.B();

 

为了给A.B() 提供扩展性， 最好的方式就是A是接口， 所有实现了A接口的类都会实现B接口，这就是RUnnale接口的由来。同理callable 接口也是一样， 只不过多了返回值。

 

谁来接收这个返回值呢， Future 接口。

 

FutureTask 是Future接口的实现， 是对callable和Runnable 接口的包装。

 

Future 接口的所有实现其实是操作callable和runnable的细节

 

package Future;

 

import java.util.concurrent.Callable;

 

public class FutureCase implements Callable{

 

/**

* @param args

*/

public static void main(String[] args) {

System.out.println("test Future");

 

}

 

@Override

public Object call() throws Exception {

// TODO Auto-generated method stub

System.out.println("test Future");

Thread.sleep(10000);

System.out.println("test Future 10000");

Thread.sleep(10000);

return "call";

}

 

}

 

package Future;

 

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.Future;

 

 

public class TestFuture {

 

/**

* @param args

*/

public static void main(String[] args) {

 

ExecutorService exec = Executors.newCachedThreadPool();    

Future<String> result = exec.submit(new FutureCase());

System.out.println("i need to do my work at first");

try {

String res =  result.get();

System.out.println("i need to do my work at second");

System.out.println(res);

} catch (InterruptedException e) {

// TODO Auto-generated catch block

e.printStackTrace();

} catch (ExecutionException e) {

// TODO Auto-generated catch block

e.printStackTrace();

}

 

}

 

}

 

 

 

 

 

FutureTask 实现。

package java.util.concurrent;

 

import java.util.concurrent.locks.LockSupport;

import sun.misc.Unsafe;

 

public class FutureTask<V>

  implements RunnableFuture<V>

{

  private volatile int state;

  private static final int NEW = 0;

  private static final int COMPLETING = 1;

  private static final int NORMAL = 2;

  private static final int EXCEPTIONAL = 3;

  private static final int CANCELLED = 4;

  private static final int INTERRUPTING = 5;

  private static final int INTERRUPTED = 6;

  private Callable<V> callable;

  private Object outcome;

  private volatile Thread runner;

  private volatile WaitNode waiters;

  private static final Unsafe UNSAFE;

  private static final long stateOffset;

  private static final long runnerOffset;

  private static final long waitersOffset;

 

  private V report(int paramInt)

    throws ExecutionException

  {

    Object localObject = this.outcome;

    if (paramInt == 2)

      return localObject;

    if (paramInt >= 4)

      throw new CancellationException();

    throw new ExecutionException((Throwable)localObject);

  }

 

  public FutureTask(Callable<V> paramCallable)

  {

    if (paramCallable == null)

      throw new NullPointerException();

    this.callable = paramCallable;

    this.state = 0;

  }

 

  public FutureTask(Runnable paramRunnable, V paramV)

  {

    this.callable = Executors.callable(paramRunnable, paramV);

    this.state = 0;

  }

 

  public boolean isCancelled()

  {

    return this.state >= 4;

  }

 

  public boolean isDone()

  {

    return this.state != 0;

  }

 

  public boolean cancel(boolean paramBoolean)

  {

    if (this.state != 0)

      return false;

    if (paramBoolean)

    {

      if (!UNSAFE.compareAndSwapInt(this, stateOffset, 0, 5))

        return false;

      Thread localThread = this.runner;

      if (localThread != null)

        localThread.interrupt();

      UNSAFE.putOrderedInt(this, stateOffset, 6);

    }

    else if (!UNSAFE.compareAndSwapInt(this, stateOffset, 0, 4))

    {

      return false;

    }

    finishCompletion();

    return true;

  }

 

  public V get()

    throws InterruptedException, ExecutionException

  {

    int i = this.state;

    if (i <= 1)

      i = awaitDone(false, 0L);

    return report(i);

  }

 

  public V get(long paramLong, TimeUnit paramTimeUnit)

    throws InterruptedException, ExecutionException, TimeoutException

  {

    if (paramTimeUnit == null)

      throw new NullPointerException();

    int i = this.state;

    if ((i <= 1) && ((i = awaitDone(true, paramTimeUnit.toNanos(paramLong))) <= 1))

      throw new TimeoutException();

    return report(i);

  }

 

  protected void done()

  {

  }

 

  protected void set(V paramV)

  {

    if (UNSAFE.compareAndSwapInt(this, stateOffset, 0, 1))

    {

      this.outcome = paramV;

      UNSAFE.putOrderedInt(this, stateOffset, 2);

      finishCompletion();

    }

  }

 

  protected void setException(Throwable paramThrowable)

  {

    if (UNSAFE.compareAndSwapInt(this, stateOffset, 0, 1))

    {

      this.outcome = paramThrowable;

      UNSAFE.putOrderedInt(this, stateOffset, 3);

      finishCompletion();

    }

  }

 

  public void run()

  {

    if ((this.state != 0) || (!UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread())))

      return;

    try

    {

      Callable localCallable = this.callable;

      if ((localCallable != null) && (this.state == 0))

      {

        Object localObject1;

        int j;

        try

        {

          localObject1 = localCallable.call();

          j = 1;

        }

        catch (Throwable localThrowable)

        {

          localObject1 = null;

          j = 0;

          setException(localThrowable);

        }

        if (j != 0)

          set(localObject1);

      }

      this.runner = null;

      int i = this.state;

      if (i >= 5)

        handlePossibleCancellationInterrupt(i);

    }

    finally

    {

      this.runner = null;

      int k = this.state;

      if (k >= 5)

        handlePossibleCancellationInterrupt(k);

    }

  }

 

  protected boolean runAndReset()

  {

    if ((this.state != 0) || (!UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread())))

      return false;

    int i = 0;

    int j = this.state;

    try

    {

      Callable localCallable = this.callable;

      if ((localCallable != null) && (j == 0))

        try

        {

          localCallable.call();

          i = 1;

        }

        catch (Throwable localThrowable)

        {

          setException(localThrowable);

        }

      this.runner = null;

      j = this.state;

      if (j >= 5)

        handlePossibleCancellationInterrupt(j);

    }

    finally

    {

      this.runner = null;

      j = this.state;

      if (j >= 5)

        handlePossibleCancellationInterrupt(j);

    }

    return (i != 0) && (j == 0);

  }

 

  private void handlePossibleCancellationInterrupt(int paramInt)

  {

    if (paramInt == 5)

      while (this.state == 5)

        Thread.yield();

  }

 

  private void finishCompletion()

  {

    Object localObject;

    while ((localObject = this.waiters) != null)

      if (UNSAFE.compareAndSwapObject(this, waitersOffset, localObject, null))

        while (true)

        {

          Thread localThread = ((WaitNode)localObject).thread;

          if (localThread != null)

          {

            ((WaitNode)localObject).thread = null;

            LockSupport.unpark(localThread);

          }

          WaitNode localWaitNode = ((WaitNode)localObject).next;

          if (localWaitNode == null)

            break;

          ((WaitNode)localObject).next = null;

          localObject = localWaitNode;

        }

    done();

    this.callable = null;

  }

 

  private int awaitDone(boolean paramBoolean, long paramLong)

    throws InterruptedException

  {

    long l = paramBoolean ? System.nanoTime() + paramLong : 0L;

    WaitNode localWaitNode = null;

    boolean bool = false;

    while (true)

    {

      if (Thread.interrupted())

      {

        removeWaiter(localWaitNode);

        throw new InterruptedException();

      }

      int i = this.state;

      if (i > 1)

      {

        if (localWaitNode != null)

          localWaitNode.thread = null;

        return i;

      }

      if (i == 1)

      {

        Thread.yield();

      }

      else if (localWaitNode == null)

      {

        localWaitNode = new WaitNode();

      }

      else if (!bool)

      {

        bool = UNSAFE.compareAndSwapObject(this, waitersOffset, localWaitNode.next = this.waiters, localWaitNode);

      }

      else if (paramBoolean)

      {

        paramLong = l - System.nanoTime();

        if (paramLong <= 0L)

        {

          removeWaiter(localWaitNode);

          return this.state;

        }

        LockSupport.parkNanos(this, paramLong);

      }

      else

      {

        LockSupport.park(this);

      }

    }

  }

 

  private void removeWaiter(WaitNode paramWaitNode)

  {

    if (paramWaitNode != null)

    {

      paramWaitNode.thread = null;

      Object localObject1 = null;

      WaitNode localWaitNode;

      for (Object localObject2 = this.waiters; ; localObject2 = localWaitNode)

      {

        if (localObject2 == null)

          return;

        localWaitNode = ((WaitNode)localObject2).next;

        if (((WaitNode)localObject2).thread != null)

        {

          localObject1 = localObject2;

        }

        else

        {

          if (localObject1 != null)

          {

            localObject1.next = localWaitNode;

            if (localObject1.thread != null)

              continue;

            break;

          }

          if (!UNSAFE.compareAndSwapObject(this, waitersOffset, localObject2, localWaitNode))

            break;

        }

      }

    }

  }

 

  static

  {

    try

    {

      UNSAFE = Unsafe.getUnsafe();

      FutureTask localFutureTask = FutureTask.class;

      stateOffset = UNSAFE.objectFieldOffset(localFutureTask.getDeclaredField("state"));

      runnerOffset = UNSAFE.objectFieldOffset(localFutureTask.getDeclaredField("runner"));

      waitersOffset = UNSAFE.objectFieldOffset(localFutureTask.getDeclaredField("waiters"));

    }

    catch (Exception localException)

    {

      throw new Error(localException);

    }

  }

 

  static final class WaitNode

  {

    volatile Thread thread = Thread.currentThread();

    volatile WaitNode next;

  }

}

 

/* Location:           C:\Program Files\Java\jre7\lib\rt.jar

 * Qualified Name:     java.util.concurrent.FutureTask

 * JD-Core Version:    0.6.2

 

 */