线程池—FutureTask源码解析

简介

在学习线程池之前，需要先了解FutureTask，因为线程池的submit方法返回的结果就是FutureTask。FutureTask表示一个未来的任务，当前调用线程会阻塞直到获取到结果。

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适用于异步获取执行结果或取消任务的场景。通过传入Runnable或Callable任务，FutureTask可以执行任务，并在外部通过get方法异步获取结果。FutureTask还能确保多次调用run方法只执行一次，或者通过cancel方法取消任务。

线程池返回的Future

线程池的submit方法返回的是FutureTask。例如：

public class FutureTaskDemo {
    public static void main(String[] args) throws Exception {
        ExecutorService threadPool = Executors.newFixedThreadPool(5);
        FutureTask future = threadPool.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println("FutureTaskDemo.run");
            }
        });
        future.get();
    }
}

开启线程的方式

通过Callable接口实现线程启动：

public static void main(String[] args) throws ExecutionException, InterruptedException {
    Callable
   
     callable = new Callable
    
     () {
        @Override
        public Integer call() throws Exception {
            return 1;
        }
    };
    FutureTask
     
       task = new FutureTask<>(callable);
    Thread thread = new Thread(task);
    thread.start();
    int x = task.get();
    System.out.println(x);
}

FutureTask源码解析

属性

state: 表示任务状态。
- NEW: 任务未执行。
- COMPLETING: 任务完成中。
- NORMAL: 任务正常完成。
- EXCEPTIONAL: 任务异常完成。
- CANCELLED: 任务被取消。
- INTERRUPTING: 任务被中断。
- INTERRUPTED: 任务中断完成。

callable: 实现Callable接口的任务。

outcome: 任务结果或异常。

runner: 执行任务的线程。

waiters: 等待任务完成的线程集合。

构造方法

FutureTask(Callable<V> callable): 传入Callable实现类。

FutureTask(Runnable runnable, V result): 通过适配器模式将Runnable转换为Callable。

run()方法

public void run() {
    if (state != NEW || 
        !UNSAFE.compareAndSwapObject(
            this, runnerOffset, 
            null, Thread.currentThread()
        )) {
        return;
    }
    try {
        if (callable != null && state == NEW) {
            Object result = null;
            boolean ran = true;
            try {
                result = callable.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                setException(ex);
            }
            if (ran) {
                set(result);
            }
        }
    } finally {
        runner = null;
        int s = state;
        if (s >= INTERRUPTING) {
            handlePossibleCancellationInterrupt(s);
        }
    }
}

get()方法

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING) {
        s = awaitDone(false, 0L);
    }
    return report(s);
}

awaitDone()方法

private int awaitDone(boolean timed, long nanos) throws InterruptedException {
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    WaitNode q = null;
    boolean queued = false;
    for (;;) {
        if (Thread.interrupted()) {
            removeWaiter(q);
            throw new InterruptedException();
        }
        int s = state;
        if (s > COMPLETING) {
            if (q != null) {
                q.thread = null;
            }
            return s;
        } else if (s == COMPLETING) {
            Thread.yield();
        } else if (q == null) {
            q = new WaitNode();
        } else if (!queued) {
            queued = UNSAFE.compareAndSwapObject(
                this, waitersOffset, q.next = waiters, q
            );
        } else if (timed) {
            nanos = deadline - System.nanoTime();
            if (nanos <= 0) {
                removeWaiter(q);
                return state;
            }
            LockSupport.parkNanos(this, nanos);
        } else {
            LockSupport.park(this);
        }
    }
}

完成线程池源码解析后，可以更好地理解线程池的工作原理。

转载地址：http://pkhfk.baihongyu.com/

你可能感兴趣的文章