Java并发之Callable,Future,FutureTask

g21121

浏览: 694631 次
性别:
来自: 北京

最近访客更多访客>>

lbyzx123

cuityang

jin361612388

knight_black_bob

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

在传统的多线程实现方式中（继承Thread和实现Runnable）无法直接获取线程执行的返回结果，如果需要获取执行结果，就必须通过共享变量或者使用线程通信的方式来达到效果，这样使用起来就比较麻烦。

从Java 1.5开始，java.util.concurrent包中提供了 Callable和 Future两个接口，通过它们就可以在任务执行完毕之后得到任务执行结果。

一、Callable

返回可能抛出异常的任务的结果。实现者定义了一个不带任何参数的叫做 call 的方法。

1.Callable与 Runnable

Callable 接口类似于 Runnable，两者都是为那些其实例可能被另一个线程执行的类设计的。但是 Runnable 不会返回结果，并且无法抛出经过检查的异常。

Callable是 java.util.concurrent中的接口，而 Runnable是 java.lang包中的接口。它们都只有一个方法。

以下是 Runnable接口的源代码：

public interface Runnable {
    /**
     * 使用实现接口 Runnable 的对象创建一个线程时，
     * 启动该线程将导致在独立执行的线程中调用对象的 run 方法。 
     */
    public abstract void run();
}

由于 run方法返回类型是 void，所以在执行完任务之后无法获得任何返回结果。

接下来是 Callable接口的源代码：

public interface Callable<V> {
    /**
     * 计算结果，如果无法计算结果，则抛出一个异常。 
     */
    V call() throws Exception;
}

可以看到 call方法返回了一个泛型V结果，所以我们就可以通过结果来进一步判断处理。

2.Callable转换

Executors 类包含一些从其他普通形式（Runnable、PrivilegedAction等）转换成 Callable 类的实用方法。

以下是 Executors类中将 Runnable转换成 Callable的相关方法：

/**
 * 返回 Callable 对象，调用它时可运行给定的任务并返回 null
 */
static Callable<Object> callable(Runnable task) 
/**
 * 返回 Callable 对象，调用它时可运行给定的任务并返回给定的结果
 */
static <T> Callable<T> callable(Runnable task, T result)

我们可以通过 Executors类的callable方法将 Runnable 转换成 Callable，其中 result为返回结果，以下是callable方法的源代码：

public static <T> Callable<T> callable(Runnable task, T result) {
    if (task == null)
        throw new NullPointerException();
    return new RunnableAdapter<T>(task, result);
}

当 task为空时会抛出空指针异常，然后通过构造方法创建一个 RunnableAdapter实例。

RunnableAdapter类是一个内部类，其实 RunnableAdapter是一个 Runnable的适配器，它的作用很简单，就是将 Runnable转换成 Callable，所以这里就用到了适配器模式，面试的时候你可以拿出去装装B。

以下是 RunnableAdapter类的源代码：

/**
 * Runnable适配器，具有返回结果的可运行任务
 */
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;
    }
}

代码即简洁有清晰，通过调用构造方法创建一个 RunnableAdapter实例，然后通过执行call方法运行线程，最后返回结果。

3.Callable 的使用

因为 Callable 接口并没有相关实现类，所以我们无法直接使用它，所以使用 Callable 需要调用 ExecutorService接口的 submit方法：

<T> Future<T> submit(Callable<T> task)

submit提交一个返回值的任务用于执行，返回一个表示任务的未决结果的 Future。该 Future 的 get 方法在成功完成时将会返回该任务的结果。

如果想立即阻塞任务的等待，则可以使用：

result = executorService.submit(aCallable).get();

注意：Executors 类包括了一组方法，可以转换某些其他常见的类似于闭包的对象，例如，将 PrivilegedAction 转换为 Callable 形式，这样就可以提交它们了。

后面的 Future章节还会更深入介绍 Callable 的相关使用。

二、Future

Future 用以表示异步计算的结果。它提供了检查计算是否完成的方法，以等待计算的完成，并获取计算的结果。计算完成后只能使用 get 方法来获取结果，如有必要，计算完成前可以阻塞此方法。取消则由 cancel 方法来执行。还提供了其他方法，以确定任务是正常完成还是被取消了。一旦计算完成，就不能再取消计算。如果为了可取消性而使用 Future 但又不提供可用的结果，则可以声明 Future<?> 形式类型、并返回 null 作为底层任务的结果。

1.继承关系

2.Future 源代码

以下是 Future接口的源代码：

public interface Future<V> {
	/**
	 * 试图取消对此任务的执行
	 */
	boolean cancel(boolean mayInterruptIfRunning);

	/**
	 * 如果在任务正常完成前将其取消，则返回 true
	 */
	boolean isCancelled();

	/**
	 * 如果任务已完成，则返回 true
	 */
	boolean isDone();

	/**
	 * 如有必要，等待计算完成，然后获取其结果
	 */
	V get() throws InterruptedException, ExecutionException;

	/**
	 * 如有必要，最多等待为使计算完成所给定的时间之后，获取其结果（如果结果可用）
	 */
	V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

3.Future 相关方法

Future包含了5个方法。

1）get()：获取其结果，这个方法会产生阻塞，会一直等到任务执行完毕才返回。

2）get(long timeout,TimeUnit unit)：与get()方法不同的是，在指定时间内等待结果的返回，如果指定时间已到仍没有结果，则返回null。

3）isDone()：如果任务已完成，则返回 true。可能由于正常终止、异常或取消而完成，在所有这些情况中，此方法也都将返回 true。

4）cancel(boolean mayInterruptIfRunning)：试图取消对此任务的执行。如果任务已完成、已取消或者由于某些其他原因而无法取消，则此尝试将失败。当调用 cancel 时，如果调用成功，而此任务尚未启动，则此任务将永不运行。如果任务已经启动，则 mayInterruptIfRunning 参数确定是否应该以试图停止任务的方式来中断执行此任务的线程。

此方法返回后，对 isDone() 的后续调用将始终返回 true。如果此方法返回 true，则对 isCancelled() 的后续调用将始终返回 true。

5）isCancelled()：如果在任务正常完成前将其取消，则返回 true。

通过 Future可以：

1）判断任务是否完成；

2）能够中断任务；

3）能够获取任务执行结果。

4.Future的使用

1）执行Future

使用 ExecutorService的 submit(Callable<T> task)方法。

public class FutureTest {

	public static void main(String[] args) throws InterruptedException, ExecutionException {
		// 创建线程池
		ExecutorService es = Executors.newCachedThreadPool();
		// 创建FutureThread实例
		FutureThread ft = new FutureThread();
		// 调用ExecutorService的 submit(Callable<T> task)方法，执行ft
		Future<Integer> f = es.submit(ft);
		// 关闭线程池
		es.shutdown();
		// 获取结果
		Integer result = f.get();
		// 判断结果
		if (result != null) {
			System.out.println("执行成功，结果为：" + result);
		} else {
			System.out.println("执行失败");
		}
	}
}

class FutureThread implements Callable<Integer> {

	// 执行函数
	public Integer call() throws Exception {
		System.out.println("开始执行");
		return new Random().nextInt(100);
	}
}
//结果：
开始执行
执行成功，结果为：51

首先，创建一个 Callable实现类，其中 call方法中是线程要执行的方法，这与 run方法类似。

然后，创建一个 ExecutorService实例，也就是线程池，然后调用submit方法将之前的 Callable实现传递执行。

最终，通过 Future的get方法可以获取到线程的执行结果，进一步处理。

2）取消 Future

使用 Future的 cancel方法尝试取消Future：

public class FutureTest {

	public static void main(String[] args) throws InterruptedException, ExecutionException {
		// 创建线程池
		ExecutorService es = Executors.newCachedThreadPool();
		// 创建FutureThread实例
		FutureThread ft = new FutureThread();
		// 调用ExecutorService的 submit(Callable<T> task)方法，执行ft
		Future<Integer> f = es.submit(ft);
		// 取消
		f.cancel(true);
		if (!f.isCancelled()) {
			// 获取结果
			Integer result = f.get();
			// 判断结果
			if (result != null) {
				System.out.println("执行成功，结果为：" + result);
			} else {
				System.out.println("执行失败");
			}
		} else {
			System.out.println("执行已被取消");
		}
		// 关闭线程池
		es.shutdown();
	}
}

class FutureThread implements Callable<Integer> {

	// 执行函数
	public Integer call() throws Exception {
		System.out.println("开始执行");
		return new Random().nextInt(100);
	}
}
//结果：
执行已被取消

3）判断Future是否完成

可以通过isDone()方法判断 Future是否已经完成：

public class FutureTest {

	public static void main(String[] args) throws InterruptedException, ExecutionException {
		// 创建线程池
		ExecutorService es = Executors.newCachedThreadPool();
		// 创建FutureThread实例
		FutureThread ft = new FutureThread();
		// 调用ExecutorService的 submit(Callable<T> task)方法，执行ft
		Future<Integer> f = es.submit(ft);
		// 取消
		// f.cancel(true);
		if (!f.isCancelled()) {
			for (;;) {
				if (f.isDone()) {
					// 获取结果
					Integer result = f.get();
					// 判断结果
					if (result != null) {
						System.out.println("执行成功，结果为：" + result);
					} else {
						System.out.println("执行失败");
					}
					break;
				} else {
					System.out.println("执行未完成");
				}
			}
		} else {
			System.out.println("执行已被取消");
		}
		// 关闭线程池
		es.shutdown();
	}
}

class FutureThread implements Callable<Integer> {

	// 执行函数
	public Integer call() throws Exception {
		System.out.println("开始执行");
		return new Random().nextInt(100);
	}
}
//结果：
执行未完成
执行未完成
开始执行
...
...
执行未完成
执行未完成
执行成功，结果为：72

当然我们并不需要这样来循环判断，因为 get方法可以阻塞等待结果。

5.RunnableFuture

RunnableFuture作为 Runnable 的 Future。成功执行 run 方法可以完成 Future 并允许访问其结果。RunnableFuture接口在 Future接口的基础上增加了一个run方法：

public interface RunnableFuture<V> extends Runnable, Future<V> {
    /**
     * 在未被取消的情况下，将此 Future 设置为计算的结果
     */
    void run();
}

注意：在 Java中class是无法多继承的，但接口可以。

6.ScheduledFuture

ScheduledFuture是一个延迟的、结果可接受的操作，可将其取消 Future。通常已安排的 future 是用 ScheduledExecutorService 安排任务的结果

ScheduledFuture接口继承自Delayed与 Future，其中没有任何独立方法：

public interface ScheduledFuture<V> extends Delayed, Future<V> {
}

7.RunnableScheduledFuture

RunnableScheduledFuture继承自 Runnable 的 ScheduledFuture。成功执行 run 方法可以完成 Future 并允许访问其结果。

RunnableScheduledFuture 增加了一个isPeriodic()方法，用来判断是否是一个定期任务：

public interface RunnableScheduledFuture<V> extends RunnableFuture<V>, ScheduledFuture<V> {

    /**
     * 如果这是一个定期任务，则返回 true。定期任务可以根据计划重新运行。非定期任务只能运行一次
     */
    boolean isPeriodic();
}

三、FutureTask

可取消的异步计算任务。FutureTask提供了对 Future 的基本实现。仅在计算完成时才能获取结果；如果计算尚未完成，则阻塞 get 方法。一旦计算完成，就不能再重新开始或取消计算。

可使用 FutureTask 包装 Callable 或 Runnable 对象。因为 FutureTask 实现了 Runnable，所以可将 FutureTask 提交给 Executor 执行。

除了作为一个独立的类外，FutureTask还提供了 protected 功能（done()、set(V v)等方法），这在创建自定义任务类时可能很有用。

1.构造方法

有两个构造方法：

/**
 * 创建一个 FutureTask，一旦运行就执行给定的 Callable
 */
public FutureTask(Callable<V> callable) {
	if (callable == null)
		throw new NullPointerException();
	sync = new Sync(callable);
}

/**
 * 创建一个 FutureTask，一旦运行就执行给定的 Runnable，并安排成功完成时 get 返回给定的结果
 */
public FutureTask(Runnable runnable, V result) {
	sync = new Sync(Executors.callable(runnable, result));
}

一种是直接指定 Callable为参数。另一种则是指定了 Runnable与 result。

两个构造方法都是创建了一个 Sync实例，最终传递的参数都为 callable，Sync内部类我们已经接触过很多次了，虽然不同类中 Sync的细节实现各不相同，但实现思想基本一致。

2.Sync内部类

Sync作为 FutureTask的内部类，用以控制 FutureTask的同步。

1）构造方法

Sync只有一个构造方法。

Sync(Callable<V> callable) {
	this.callable = callable;
}

2）成员变量

/** 表示任务正在运行的状态值 */
private static final int RUNNING = 1;
/** 表示任务运行完成的状态值 */
private static final int RAN = 2;
/** 表示任务已被取消的状态值 */
private static final int CANCELLED = 4;

/** 底层 callable */
private final Callable<V> callable;
/** 返回结果，通过get方法获取 */
private V result;
/** 从get方法中抛出的异常 */
private Throwable exception;

/**
 * 运行任务的线程
 */
private volatile Thread runner;

从成员变量中可以看到 FutureTask拥有四种状态：未运行（0）、运行中（1），已运行（2）和已取消（4）。其中未运行是默认状态，状态值为0，其他三种已经在类中定义。其他几个成员变量分别是运行相关的声明。

3）核心方法

（1）innerGet方法

innerGet方法是Sync中获取执行结果的方法，FutureTask的 get方法也是调用的本方法。

以下是 innerGet方法的源代码：

V innerGet() throws InterruptedException, ExecutionException {
	acquireSharedInterruptibly(0);
	if (getState() == CANCELLED)
		throw new CancellationException();
	if (exception != null)
		throw new ExecutionException(exception);
	return result;
}

innerGet方法首先调用的是AQS（AbstractQueuedSynchronizer）的 acquireSharedInterruptibly方法，最终调用的是 doAcquireSharedInterruptibly方法。然后判断任务状态，如果已经取消则抛出 CancellationException异常，然后判断异常信息，如果有则抛出。最后返回结果。

doAcquireSharedInterruptibly的核心代码为：

for (;;) {
    final Node p = node.predecessor();
    if (p == head) {
        int r = tryAcquireShared(arg);
        if (r >= 0) {
            setHeadAndPropagate(node, r);
            p.next = null; // help GC
            return;
        }
    }
    if (shouldParkAfterFailedAcquire(p, node) &&
        parkAndCheckInterrupt())
        break;
}

（2）innerGet(long nanosTimeout)方法

与 innerGet()方法类似，只是innerGet(long nanosTimeout)会在指定时间内未得到结果时返回null。

（3）innerRun()方法

FutureTask运行任务时会调用 Sync的 innerRun方法。

void innerRun() {
	//首先设置状态，未运行-->运行中
    if (!compareAndSetState(0, RUNNING))
        return;
    try {
    	//获取当前线程
        runner = Thread.currentThread();
        //判断状态
        if (getState() == RUNNING)
            //开始执行
             innerSet(callable.call());
        else
            //释放共享锁
            releaseShared(0);
    } catch (Throwable ex) {
    	//处理异常
        innerSetException(ex);
    }
}

首先会将未运行状态设置成运行中，也就是判断该任务是否已经执行过，然后再设置状态。其中会调用 innerSet方法来执行任务。调用 releaseShared方法来取消任务。

（4）innerSet(V v)方法

void innerSet(V v) {
	for (;;) {
		//获取状态
		int s = getState();
		//如果s==2，即已运行过，则返回
		if (s == RAN)
			return;
		//如果s==4，即已取消
		if (s == CANCELLED) {
			//释放共享锁
			releaseShared(0);
			return;
		}
		//设置状态为RAN，值为2
		if (compareAndSetState(s, RAN)) {
                        //设置结果
			result = v;
			//释放共享锁
			releaseShared(0);
			//调用done方法，done方法需要使用者自行实现
			done();
			return;
		}
	}
}

（5）innerSetException

innerSetException方法与innerSet方法流程基本一致，只是在其中添加了异常信息的设置。

void innerSetException(Throwable t) {
	for (;;) {
		//获取状态
		int s = getState();
		//如果已运行则返回
		if (s == RAN)
			return;
		//如果为取消状态
		if (s == CANCELLED) {
			//释放共享锁
			releaseShared(0);
			return;
		}
		//设置状态为RAN
		if (compareAndSetState(s, RAN)) {
			//设置异常信息
			exception = t;
			result = null;
			//释放共享锁
			releaseShared(0);
			done();
			return;
		}
	}
}

（6）innerCancel

innerCancel方法用于取消任务执行。

boolean innerCancel(boolean mayInterruptIfRunning) {
	for (;;) {
		// 获取状态
		int s = getState();
		// 判断状态是否为 RAN或 CANCELLED
		if (ranOrCancelled(s))
			return false;
		// 设置状态值为 CANCELLED
		if (compareAndSetState(s, CANCELLED))
			break;
	}
	// 如果可以中断
	if (mayInterruptIfRunning) {
		Thread r = runner;
		if (r != null)
			// 中断线程
			r.interrupt();
	}
	// 释放共享锁
	releaseShared(0);
	//调用done方法
	done();
	return true;
}

（7）innerRunAndReset

innerRunAndReset与innerRun不同之处在于，innerRunAndReset只运行却不返回结果，所以innerRunAndReset更适合一些重置恢复的操作。

boolean innerRunAndReset() {
	// 将状态为未运行设置为运行中
	if (!compareAndSetState(0, RUNNING))
		return false;
	try {
		runner = Thread.currentThread();
		// 再次判断状态
		if (getState() == RUNNING)
			// 执行call
			callable.call(); // 不设置结果
		runner = null;
		// 将状态重置为未运行
		return compareAndSetState(RUNNING, 0);
	} catch (Throwable ex) {
		// 设置异常
		innerSetException(ex);
		return false;
	}
}

4）其他方法

（1）ranOrCancelled

判断状态是否为 RAN或 CANCELLED状态。

private boolean ranOrCancelled(int state) {
	return (state & (RAN | CANCELLED)) != 0;
}

可以看到 ranOrCancelled是通过状态值与 RAN或 CANCELLED值做与运算。

假设此时state值为0，与运算后结果为0，则最终结果就为false，所以即不为 RAN也不为 CANCELLED。

（2）innerIsDone

判断内部执行是否完成。

boolean innerIsDone() {
	return ranOrCancelled(getState()) && runner == null;
}

首先判断状态是否为 RAN或 CANCELLED，然后再判断 runner是否为空。

（3）tryReleaseShared

尝试释放共享锁。

/**
 * AQS的基本实现，当状态为已完成和runner为空时释放共享锁
 */
protected boolean tryReleaseShared(int ignore) {
	runner = null;
	return true;
}

可以看到 tryReleaseShared为 protected 方法，而且方法体中也没有太具体的实现，那是因为这种 protected 修饰的方法需要使用者自行实现。

还有几个其他类似的方法，这里就不一一说明了，它们的原理跟这几个方法类似。

3.FutureTask方法

1）get()方法

等待计算完成，然后获取其结果，如果未返回结果则阻塞。

以下是get方法的源代码：

public V get() throws InterruptedException, ExecutionException {
	return sync.innerGet();
}

get方法调用的是Sync内部类的 innerGet方法，innerGet方法在上面已经分析过，这里就不赘述了。

2）get(long timeout, TimeUnit unit)方法

在指定时间内等待，如果时间已到无结果则返回null。

3）set(V v)方法

除非已经设置了此 Future 或已将其取消，否则将其结果设置为给定的值。在计算成功完成时通过 run 方法内部调用此方法。

set方法调用的是Sync内部类的innerSet方法。

protected void set(V v) {
    sync.innerSet(v);
}

4）run()方法

除非已将此 Future 取消，否则将其设置为其计算的结果。

run方法调用的是Sync内部类的 innerRun方法。

public void run() {
	sync.innerRun();
}

5）runAndReset()方法

执行计算而不设置其结果，然后将此 Future 重置为初始状态，如果计算遇到异常或已取消，则该操作失败。本操作被设计用于那些本质上要执行多次的任务。

runAndReset方法调用的是Sync内部类的 runAndReset方法。

protected boolean runAndReset() {
    return sync.innerRunAndReset();
}

6）cancel(boolean mayInterruptIfRunning)方法

试图取消对此任务的执行。如果任务已完成、或已取消，或者由于某些其他原因而无法取消，则此尝试将失败。当调用 cancel 时，如果调用成功，而此任务尚未启动，则此任务将永不运行。如果任务已经启动，则 mayInterruptIfRunning 参数确定是否应该以试图停止任务的方式来中断执行此任务的线程。

cancel方法调用的也是Sync内部类的 innerCancel方法。

public boolean cancel(boolean mayInterruptIfRunning) {
    return sync.innerCancel(mayInterruptIfRunning);
}

可以看到 FutureTask的几个方法其实调用的都是Sync内部类的相关方法，所以 FutureTask的其他方法就不介绍了，只需要把 Sync的相关方法理解就可以了。

4.FutureTask使用

FutureTask的各个方法我们基本已经了解了，接下来就是通过实例来学习如何使用 FutureTask。

FutureTask与 Future的使用方式类似：

public class FutureTest {

	public static void main(String[] args) throws InterruptedException, ExecutionException {
		// 创建线程池
		ExecutorService es = Executors.newCachedThreadPool();
		// 创建FutureThread实例
		FutureThread ft = new FutureThread();

		FutureTask<Integer> futureTask = new FutureTask<Integer>(ft);
		// 第一种方式，使用方式与 Future相同
		es.submit(futureTask);

		// 第二种方式，使用 Thread运行 Runnable实现，同样可以得到结果
		//Thread thread = new Thread(futureTask);
		//thread.start();

		// 关闭线程池
		es.shutdown();
		// 获取结果
		Integer result = futureTask.get();
		// 判断结果
		if (result != null) {
			System.out.println("执行成功，结果为：" + result);
		} else {
			System.out.println("执行失败");
		}
	}
}

class FutureThread implements Callable<Integer> {

	// 执行函数
	public Integer call() throws Exception {
		System.out.println("开始执行");
		return new Random().nextInt(100);
	}
}

除了之上创建FutureTask实例的方式外，我们还可以这样来写：

FutureTask<Integer> futureTask = new FutureTask<Integer>(new Callable<Integer>() {
	public Integer call() throws Exception {
		System.out.println("开始执行");
		return new Random().nextInt(100);
	}
});

1
顶

6
踩

分享到：

Java并发之Executor | tomcat优化

2016-05-21 12:42
浏览 5328
评论(0)
分类:编程语言
查看更多

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论