Runable和Callable的区别？你必须要搞清楚Thread以及FutureTask！

妖狐艹你老母 2024-03-26 16:31 13阅读 0赞

Runable与Callable的区别，据说是高频面试题，什么样的答案才会让面试官满意呢？

所有java程序员都知道的答案是基于：

public interface Runnable {
        
        public abstract void run();
    }

public interface Callable<V> {
        V call() throws Exception;
    }

从Runable和Callable接口的定义我们就能知道的是：

***Runable接口的run方法不返回结果，不允许他的实现类抛出异常。***

***Callable接口的call方法可以返回结果，并且允许抛出异常。***

再和Thread关联一下，可以补充：

1.  新创建一个线程Thread执行Runable可以实现异步调用，调用线程在创建新线程执行Runable任务之后可以立即获得执行权，不影响后续任务的执行
2.  Callable可以被封装在FutureTask（Runable的实现类）中从而被Thread调用，这种情况下调用线程可以通过FutureTask.get()方法获取到Callable接口的call方法的返回值，但是FutureTask.get()方法是阻塞调用，直到线程执行完成才会有结果返回，所以调用线程如果执行了FutureTask.get()的话，后续任务就必须等待

如果单从这个问题来讲，答案应该就差不多完整了。但是如果面试官老爷还想了解细节，那我们就必须要对FutureTask甚至Executor做一个深入的了解。

其实我们并不是为了回答这个面试问题才去研究这部分内容的，而是因为接下来准备认真学习、研究一下线程池、连接池等池化技术，而FutureTask、Executor是线程池技术的基础，所以我们就需要首先搞清楚FutureTask和Executor的底层原理，顺道也许正好能把这个高频java基础面试题回答清楚。

进入正题。今天的内容包括：

1.  Thread：概念性的简单说明，不涉及启动、停止、中断、挂起等等状态转换机制
2.  FutureTask源码分析

Thread简要说明

Java程序是以线程来运行的，JVM允许一个java应用启动多个线程，当我们的程序以main方法启动之后，其实JVM就启动了一个线程来执行程序。

每一个java thread都有优先级，在资源争夺过程中高优先级线程比低优先级线程更容易获得执行权。线程可以被标注为是否为守护线程，默认情况下，新创建的线程和创建他的线程拥有相同的优先级，当前仅当创建线程是守护线程的时候，被创建的新线程才能是守护线程。

JVM启动的时候，通常会创建一个非守护线程（主线程），主线程会持续运行直到：

1.  Runtime的exit方法执行，或者应用抛出的异常一直被传递到主线程的run方法
2.  所有非守护线程执行完毕：run方法正常执行完成或捕获到异常

有两种创建线程的方式：

1.  创建一个Thread类的子类并覆盖他的run方法，然后调用子类对象的start方法，最终子类的run方法会被调用执行
2.  创建Runable接口的实现类，作为参数传递给Thread的构造器创建Thread并执行start方法，最终Runable实现类的run方法会被调用执行

常见的是第2种方法。

FutureTask

先看一下FutureTask的类图：

![784c1f5b877d0965d17b823176adb8f8.png][]

第一点需要知道的是，FutureTask是Runable接口的实现。

同时他还实现了Future接口，所以我们需要首先简单了解一下Future，看一下JavaDoc：

> 体现异步任务的执行结果，提供方法检查任务是否完成、等待异步任务执行完成、并获取执行结果。执行结果只能通过get方法获取，只有任务完成后才能获取到，否则会阻塞等待任务完成后返回结果。通过cancel方法可以取消任务，Future还提供了一些其他方法可以帮助获取到任务是正常结束、还是被取消掉了，一旦任务执行完成，则不再允许取消。如果你只是想通过Future来实现任务可以被取消这一特性、而并不在意任务的返回结果，让Future的任务返回null即可。

所以我们知道Future是为了获取到异步任务的执行结果而生的。

FutureTask的任务

FutureTask包含一个叫callable的Callable成员变量，就是FutureTask的底层任务，我们首先看一下这个底层任务是怎么被创建或者初始化的。

先来看FutureTask对象的创建，FutureTask提供了两个构造方法：

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) {
            this.callable = Executors.callable(runnable, result);
            this.state = NEW;       // ensure visibility of callable
        }

其中一个构造方法传入Callable对象，直接赋值给成员变量callable，设置state=NEW，简单明了。

另外一个构造方法传入Runable对象，以及一个result对象，FutureTask首先调用Executors的callable方法把传入的Runable对象包装成一个Callable对象，然后把包装好的Callable对象赋值给成员变量callable，设置state=NEW，结束。

***所以我们知道，Future的底层任务只能是Callable对象，但是他也可以接收Runable对象，他自己内部负责将Runable封装成Callable，对调用方来说是透明的。***

Runable封装为Callable

Runable是通过Executors的callable方法封装为Callable的，最终生成了一个RunnableAdapter对象：

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实现了Callable接口，将Callable的call方法调用转换为对Runable对象的run方法调用，call方法的返回结果实际上是调用方传进来的。

Runable对象被这样封装之后，也就具备了通过Thread异步调用之后可以通过Future的get方法阻塞获取返回的特性了，我们需要知道，阻塞是真的，这个返回结果result其实是你在创建FutureTask对象的时候传进去的那个result，任务执行完成后原样返回。

FutureTask的状态

FutureTask的状态state：

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;

对象创建后状态为NEW，任务被线程调起后变更为COMPLETING，任务正常执行完成后为NORMAL，执行过程中抛出异常为EXCEPTIONAL，任务被取消为CANCELLED，任务被中断为INTERRUPTING->INTERRUPTED。

FutureTask的线程安全性

我们需要注意一点，FutureTask位于java.util.concurrent包下，通过UNSAFE的CAS操作以及自旋来确保线程安全线。

提前了解一下FutureTask的另外一个成员变量runner，其实就是执行callable任务的线程，初始化为null，FutureTask构造函数也没有初始化过，所以FutureTask对象创建之后应该也是null：

private volatile Thread runner;

我们在进行后续代码分析之前需要首先了解一下FutureTask定义的几个UNSAFE对象：

private static final sun.misc.Unsafe UNSAFE;
        private static final long stateOffset;
        private static final long runnerOffset;
        private static final long waitersOffset;
        static {
            try {
                UNSAFE = sun.misc.Unsafe.getUnsafe();
                Class<?> k = FutureTask.class;
                stateOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("state"));
                runnerOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("runner"));
                waitersOffset = UNSAFE.objectFieldOffset
                    (k.getDeclaredField("waiters"));
            } catch (Exception e) {
                throw new Error(e);
            }
        }

stateOffset：对应状态state的内存地址  
runnerOffset：对应线程runner的内存地址  
waitersOffset：对应阻塞对象waiters的内存地址

伪代码Demo

为了方便描述，我们以下面的伪代码举例：

//假设MyCallable是我们自己的任务
        Callable callable=new MyCallable();
        FutureTask f1=new FutureTask(callable);
        Thread t1= new Thread(f1);
        t1.start();
        Object o = f1.get();
        ...

其中MyCallable就是我们自己的Callable实现类，篇幅关系，就不贴出所有的代码了。

这个例子中会涉及到两个线程，一个是主线程我们称之为tmain，另外一个是执行FutureTask f1的线程我们称之为t1。

FutureTask的run方法

通过t1.start启动线程t1后，FutureTask的run方法会被调用。

FutureTask的run方法代码其实不长，但是为了容易理解，我们还是分段分析：

if (state != NEW ||
                !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                             null, Thread.currentThread()))
                return;

首先是判断当前状态如果不等于NEW则返回，第二句UNSAFE.compareAndSwapObject方法的意思是：直接通过JVM查看runner是否是null，如果是null的话赋值为当前Thread，不是null则直接返回。

这个时候runner就是任务执行线程t1了。

try {
                Callable<V> c = callable;
                if (c != null && state == NEW) {
                    V result;
                    boolean ran;
                    try {
                        result = c.call();
                        ran = true;
                    } catch (Throwable ex) {
                        result = null;
                        ran = false;
                        setException(ex);
                    }
                    if (ran)
                        set(result);
                }

然后通过调用callable的call方法执行任务，返回赋给result对象。如果执行成功的话，调用set方法，发生异常则调用setException(ex)进行异常处理。

我们重点看一下set(result)方法：

protected void set(V v) {
            if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
                outcome = v;
                UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
                finishCompletion();
            }
        }

状态state首先修改为COMPLETING，执行结果result赋值给outcome，为get方法的返回做准备。之后state修改为最终状态NORMAL，最后调用finishCompletion()方法。

FutureTask\#get方法

get方法是FutureTask的灵魂，如果任务尚未执行完成则get方法一直阻塞直到完成。

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

如果状态小于等于COMPLETING则调用awaitDone方法，否则调用report方法返回结果。

那一定是awaitDone方法实现阻塞等待的。

FutureTask\#awaitDone方法

> Awaits completion or aborts on interrupt or timeout.  
> 等待任务完成，或者由于中断、超时后放弃

首先for循环自旋：

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();
                }

如果当前线程被中断，从watiers队列释放本次的WaitNode、抛出异常。

判断当前状态如果大于COMPLETING则释放当前等待队列q对应的Thread后，返回，否则如果状态等于COMPLETING，说明任务还没有完成（没到最终状态NORMAL），通过调用Thread.yield()继续等待。否则，state应该就小于COMPLETING，说明任务还在执行中，如果排队队列q为null的话创建队列q=new WaitNode();否则，如果尚未排队（!queued）则q加入waiters队列。否则，调用LockSupport.part方法挂起当前线程等待唤醒。

int s = state;
                if (s > COMPLETING) {
                    if (q != null)
                        q.thread = null;
                    return s;
                }
                else if (s == COMPLETING) // cannot time out yet
                    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 <= 0L) {
                        removeWaiter(q);
                        return state;
                    }
                    LockSupport.parkNanos(this, nanos);
                }
                else
                    LockSupport.park(this);
            }

代码show完了，我们接下来分析一下这段代码的逻辑：  
第一步，我们需要首先明确一下这段代码执行的时候的“当前线程”是谁？

很明显，当前线程不是t1,而是我们上面示例代码的主线程tmain。

第二步，如果当前线程被中断，则从waiters中移出本次加入的WaitNode后抛出中断异常。

否则，如果状态变大于COMPLETING，则将当前WaitNode的线程置空直接返回。

否则如果状态为COMPLETING，则表明任务执行线程已经完成任务的执行、FutureTask尚未完成清场处理所以尚未达到最终状态，则调用Thread.yield();继续等待。

否则，t1线程还没有执行完成任务FutureTask f1，创建一个等待对象WaitNode，WaitNode对象会持有当前线程tmain：

static final class WaitNode {
            volatile Thread thread;
            volatile WaitNode next;
            WaitNode() { thread = Thread.currentThread(); }
        }

第三步，继续自旋，如果FutureTask状态仍然小于COMPLETING，则将WaitNode加入到等待队列waiters中。

第四步，继续自旋，如果FutureTask状态仍然小于COMPLETING，调用LockSupport.park(this);交出tmain线程的控制权，我们知道park方法后当前线程tmain会被挂起（处于Waiting状态或sleep状态），只能通过其他线程调用当前线程的unpark才会被唤醒。

***这样我们就很清楚的知道了get方法的逻辑：***

1.  如果t1线程已经执行完成并且FutureTask已经到了最终状态，get方法可以立刻返回
2.  否则，如果t1线程已经执行完成但是FutureTask还没有到最终状态（FutureTask的扫尾工作没完成），通过调用Thread.yield方法等待t1线程将FutureTask执行到最终状态
3.  否则，t1线程没有执行完成，则创建WaitNode对象持有当前线程，并将WaitNode对象加入waiters队列；通过LockSupport.park将调用线程tmain挂起，等待唤醒

***我们现在的问题是要找到tmain怎么被唤醒？***

FutureTask\#finishCompletion方法

我们先看一下finishCompletion的代码：

private void finishCompletion() {
            // assert state > COMPLETING;
            for (WaitNode q; (q = waiters) != null;) {
                if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
                    for (;;) {
                        Thread t = q.thread;
                        if (t != null) {
                            q.thread = null;
                            LockSupport.unpark(t);
                        }
                        WaitNode next = q.next;
                        if (next == null)
                            break;
                        q.next = null; // unlink to help gc
                        q = next;
                    }
                    break;
                }
            }
    
            done();
    
            callable = null;        // to reduce footprint
        }

从源码可以看出finishCompletion方法的逻辑：

1.  通过原子操作从waiters中获取WaitNode q
2.  ***从q中获取其持有的线程t，如果该线程不空的话，调用LockSupport.unpark(t)唤醒该线程***
3.  q的下一个节点next为空的话结束循环
4.  否则，q指向下一个节点

***关键在第2步，大概的意思是说，我执行完成之后，所有那些由于等待我执行完成而自动挂起自己的哥们，都可以被唤醒了。***

接下来就是需要验证一下finishCompletion方法在什么地方被调用：一个是cancel方法，通过FutureTask取消任务；另外一个是set（V result）方法,在run方法成功执行完成后；还有一个是setException方法，在run方法执行异常的时候。

***真相大白！***

小结

FutureTask的主要代码逻辑就分析完成了，对于FutureTask的底层逻辑应该也就有了一个深入的了解了。这个时候如果回过头再来看Runable和Callable的区别的话，应该就会有更清晰的认识了。

[784c1f5b877d0965d17b823176adb8f8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/26/b2690d85c5774998afdc5b1ba84b9896.png