java多线程编程之Future/FutureTask和Callable

有这样一种场景,用多线程发送数据到某个服务器,需要知道各个线程是否都发送成功,等所有线程都发送完成才能继续下一轮计算和发送。如果用传统的多线程方式,就需要启动多个线程,然后在每个线程中分别发送数据,外部通过某种方式等待各个线程全部都发送完成,再进行后面的计算等流程。这种实现方式的代码会比较臃肿,在java中提供了一种Callable+Future的方法,可以将异步的多线程调用变为同步方式。

Callable
在java的多线程编程中,有Thread和Runnable两种方式来新建线程,其中Runnable封装了一个异步运行的任务,可以认为是一个没有任何参数和返回值的异步方法。Callable接口类似于Runnable,两者都是为那些其实例可能被另一个线程执行的类设计的,不同之处在于:

Runnable不会返回结果,并且无法抛出经过检查的异常。而Callable是有返回结果并且可能抛出异常的。
Runnable定义了run方法,而Callable定义了一个不带任何参数的叫做call的方法。
此外,Callable接口的类型参数也是返回值的类型。

public interface Callable {
    /**
    * Computes a result, or throws an exception if unable to do so.
    *
    * @return computed result
    * @throws Exception if unable to compute a result
    */
    V call() throws Exception;
}

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

当使用Future对象时,你就可以启动一个计算,把计算结果给某线程,然后就去干自己的事。Future对象的所有者在结果计算好之后就可以得到它。

public interface Future {
boolean cancel(boolean mayInterruptIfRunning);
boolean isCancelled();
boolean isDone();
V get() throws InterruptedException, ExecutionException;
V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
}

FutureTask
FutureTask包装器是一种很方便地将Callable转换为Future和Runnable的机制,它同时实现了两者的接口。例如:

Callable comput = …;
FutureTask task = new FutureTask(comput);
Thread t = new Thread(task); // it's a Runnable
t.start();
…

Integer result = t.get(); // it’s a Future
所以可使用FutureTask包装Callable或Runnable对象。由于FutureTask实现了Runnable,可将 FutureTask提交给线程池的执行器类Executor执行。

使用Future+Callable+ExecutorService的例子

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;

public class FutureTest {

  public static class Task implements Callable {
    private long id;
    public Task(long id){
      this.id = id;
    }

    @Override
    public String call() throws Exception {
      System.out.printf("Thread#%s : in call\n", this.id);
      return this.id + "";
    }
 }

  public static void main(String[] args) throws InterruptedException, ExecutionException {
  List results = new ArrayList();
  ExecutorService execService = Executors.newCachedThreadPool();
    for(int i=0; i<10;i++)
      results.add(execService.submit(new Task(i)));

  for(Future res : results)
    System.out.println(res.get());
  }
}

可见,使用Future+Callable模式对于某些需要等待多个线程执行结果的场景非常有用,在HBase的batch操作中就使用了这种模式。

发表评论

电子邮件地址不会被公开。 必填项已用*标注