当前位置：网站首页>AQS Synchronization Component - FutureTask Analysis and Use Cases

AQS Synchronization Component - FutureTask Analysis and Use Cases

2022-08-09 13:02:00 【51CTO】

FutureTask原理

AQS同步组件-FutureTask解析和用例_线程中断
FutureTask间接实现了runnable接口和future接口,说明了futureTask是runnable与callnable的集合体,即是有返回值的runnable方法.因此,FutureTask可以交给Executor执行,也可以由调用线程直接执行（FutureTask.run()）.

源码分析

构造函数

	
	public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        this.callable = callable;
        this.state = NEW;       // ensure visibility of callable
    }

    public FutureTask(Runnable runnable, V result) {
        //When the parameter passed to the constructor is Runnable,会通过Executors.callable方法将其转换成Callable
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
    }

     1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.

FutureTaskNew provides two constructors to encapsulateCallable和Runnable,When the parameter passed to the constructor is Runnable,会通过Executors.callable方法将其转换成Callable.

常用方法

/** * 可能的状态转换:: * 新建 -> 已完成 -> 正常 * 新建 -> 已完成 -> 异常 * 新建 -> 已取消 * 新建 -> 中断ing -> 已中断 */
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;

boolean cancel(boolean mayInterruptIfRunning);//取消任务
boolean isCancelled();//是否被取消
boolean isDone();//计算是否完成
//get方法,获取执行结果,If the current thread has not finished executing, get方法会被阻塞.
public V get()
//You can set a timeout and get the execution result,If the current thread has not finished executing, get方法会被阻塞.
public V get(long timeout, TimeUnit unit)
    
/** * awaitDoneThe method is actually an endless loop,直到taskThe status changes to Completed or the wait time exceeds *Timeout or thread interruption will only jump out of the loop,程序结束; *为了节省开销,Threads don't keep spinning and waiting,而是会阻塞,使用LockSupport的parkA series of methods implement thread blocking */
private int awaitDone(boolean timed, long nanos)
    throws InterruptedException {
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    WaitNode q = null;
    boolean queued = false;
    for (;;) {
        //如果线程中断,Removes the current thread from the wait queuewaiters中移除,抛出中断异常
        if (Thread.interrupted()) {
            removeWaiter(q);
            throw new InterruptedException();
        }

        int s = state;
        //if the thread has completed,设为null
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            return s;
        }
        //如果正在执行,让出cpu
        else if (s == COMPLETING) // cannot time out yet
            Thread.yield();
       	//如果节点为空,则初始化节点
        else if (q == null)
            q = new WaitNode();
        else if (!queued)
            //CAS
            queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                 q.next = waiters, q);
        //The timeout removes the node from the queue.否则阻塞到超时.
        else if (timed) {
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) {
                removeWaiter(q);
                return state;
            }
            LockSupport.parkNanos(this, nanos);
        }
        else
            //阻塞自己
            LockSupport.park(this);
    }
}

     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.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.
67.
68.
69.
70.
71.
72.

根据FutureTask的run方法执行的时机,FutureTask可以处于以下三种执行状态：

未启动：在FutureTask.run()还没执行之前,FutureTask处于未启动状态.当创建一个FutureTask对象,并且run()方法未执行之前,FutureTask处于未启动状态.
已启动：FutureTask对象的run方法启动并执行的过程中,FutureTask处于已启动状态.
已完成：FutureTask正常执行结束,或者FutureTask执行被取消(FutureTask对象cancel方法),或者FutureTask对象run方法执行抛出异常而导致中断而结束,FutureTask都处于已完成状态.

AQS同步组件-FutureTask解析和用例_futureTask_02

当FutureTask处于未启动或已启动状态时,执行FutureTask.get()方法将导致调用线程阻塞
当FutureTask处于已完成状态时,执行FutureTask.get()方法将导致调用线程立即返回结果或抛出异常
当FutureTask处于未启动状态时,执行FutureTask.cancel()方法将导致此任务永远不会被执行
当FutureTask处于已启动状态时,执行FutureTask.cancel（true）方法将以中断执行此任务线程的方式来试图停止任务
当FutureTask处于已启动状态时,执行FutureTask.cancel（false）方法将不会对正在执行此任务的线程产生影响（让正在执行的任务运行完成）
当FutureTask处于已完成状态时,执行FutureTask.cancel（…）方法将返回false.

使用案例

FutureTask、Runnable、Callable

public static void main(String[] args) throws Exception {
        FutureTask<String> futureTask = new FutureTask<String>(new Callable<String>() {
            @Override
            public String call() throws Exception {
                log.info("callableTime-consuming task starts");
                //耗时任务
                Thread.sleep(5000);
                log.info("callable耗时任务完成");
                return "耗时任务：报告！我已完成";
            }
        });

        ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.submit(futureTask);
   		//executor.execute(futureTask);
        log.info("The main thread task starts");
        Thread.sleep(1000);
        log.info("Main thread task completed");
        log.info("Wait for the time-consuming task to complete...");
        //Get the return result of the time-consuming task,如果未返回,主线程将阻塞,处于等待状态
        String result = futureTask.get();
        log.info("result：{}", result);
    }

     1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.

输出结果如下：

22:06.250 [main] INFO com.zjq.aqs.FutureTaskExample - The main thread task starts
22:06.250 [Thread-0] INFO com.zjq.aqs.FutureTaskExample - callableTime-consuming task starts
22:07.254 [main] INFO com.zjq.aqs.FutureTaskExample - Main thread task completed
22:07.254 [main] INFO com.zjq.aqs.FutureTaskExample - Wait for the time-consuming task to complete...
22:11.254 [Thread-0] INFO com.zjq.aqs.FutureTaskExample - callable耗时任务完成
22:11.254 [main] INFO com.zjq.aqs.FutureTaskExample - result：耗时任务：报告！我已完成








可以把FutureTask交给Executor执行
也可以通过ExecutorService.submit（…）方法返回一个FutureTask,然后执行FutureTask.get()方法或FutureTask.cancel（…）方法
除此以外,还可以单独使用FutureTask

Future、Callable

public static void main(String[] args) throws Exception {

        ExecutorService executor = Executors.newCachedThreadPool();
        Future<String> future = executor.submit(new MyCallable());
        log.info("The main thread task starts");
        Thread.sleep(1000);
        log.info("Main thread task completed");
        log.info("Wait for the time-consuming task to complete...");
        //Get the return result of the time-consuming task,如果未返回,主线程将阻塞,处于等待状态
        String result = future.get();
        log.info("result：{}", result);
    }

    static class MyCallable implements Callable<String> {

        @Override
        public String call() throws Exception {
            log.info("callableTime-consuming task starts");
            //耗时任务
            Thread.sleep(5000);
            log.info("callable耗时任务完成");
            return "耗时任务：报告！我已完成";
        }
    }

     1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.