基础 | 并发编程 - [Callable & FutureTask]

§1 开线程方式

• 继承 Thread

  new Thread(()->{
     // do
  },"A").start();
  1
  2
  3

• 实现 Runable

  new Thread(()->{
     // do
  },"A").start();
  1
  2
  3

• 实现 Callable

  class Job implements Runnable{
      @Override
      public void run() {
  		// do
      }
  }
  1
  2
  3
  4
  5
  6

  与 Runable 的区别
  实现 call() 而不是 run()
  有返回值
  会抛出异常

• 实现 线程池

§2 Callable & FutureTask 使用

public class CallableJob implements Callable<String> {
    @Override
    public String call() throws Exception {
        System.out.println("=========");
        return "used";
    }

    public static void main(String[] args) {
        new Thread(new FutureTask<String>(new CallableJob())).start();
    }
}
1
2
3
4
5
6
7
8
9
10
11

FutureTask 提供了 Callable 到 Runable 的适配（适配器模式）
通过此方式解决通过开启线程调用 Callable.call() 的目的
具体方式为：
实现了 Runable 接口，并且提供一个接受 Callable 的构造方法

//public interface RunnableFuture extends Runnable, Future 
public class FutureTask<V> implements RunnableFuture<V> {
	public FutureTask(Callable<V> callable) {
	    if (callable == null)
	        throw new NullPointerException();
	    this.callable = callable;
	    this.state = NEW; 
	}
	//FutureTask 中的 run()
	public void run() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    result = c.call();// 在 run() 中调用 Callable 的 call()
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex);
                }
                if (ran)
                    set(result);
            }
        } finally {
			// ...
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

应用场景
相对于 Runable，Callable 提供一个返回值
并在结合 FutureTask 后，可以稍后获取到这个返回值
常用于分支-合并模式
即，在分步骤业务中，另开线程处理复杂或耗时的环节，并最终汇总结果

public class CallableJob implements Callable<Integer> {
    @Override
    public Integer call()  {
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 100 * 1000;
    }

    public static Integer smallJob(Integer... eles){
        return Arrays.stream(eles).mapToInt(Integer::intValue).sum();
    }

    public static void main(String[] args) {
        FutureTask<Integer> bigJob = new FutureTask<Integer>(new CallableJob());
        try {
            new Thread(bigJob).start();
            int result = smallJob(1,2,3,4,5);
            result += bigJob.get();
            System.out.println(result);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

注意

• FutureTask.get() 时，会 阻塞线程
• 可以通过设置超时时间进行控制，时间超出后会抛出 TimeoutException
• 也可是使用 FutureTask.isDone() 循环等待完成，但依然阻塞

while (!bigJob.isDone()){
    System.out.println("working");
    TimeUnit.MILLISECONDS.sleep(500);
}
result += bigJob.get();
1
2
3
4
5

• 多个线程争抢一个 FutureTask 时，只会有一个线程成功，FutureTask 只会执行一次\