FutureTask 源码阅读使用Callable创建多线程

秒速五厘米 2022-10-23 12:48 200阅读 0赞

接上文[线程池源码阅读][Link 1]线程池调用Future<?> submit(Runnable task)；返回一个FutureTsk。

## 简单使用 ##

public static void main(String[] args) throws ExecutionException, InterruptedException { 
            FutureTask<Integer> future = new FutureTask(new MyCallable());
            Thread thread = new Thread(future);
            thread.start();
            //调用get方法等待其返回（方法会阻塞线程）
            Integer result = future.get();
            System.out.println("线程运行结果" + result);
    
        }
        static class MyCallable implements Callable<Integer> { 
            @Override
            public Integer call() throws InterruptedException { 
                int i = 0;
                int sum = 0;
                while (i < 10) { 
                    sum += i;
                    i++;
                }
                return sum;
            }
        }

### 控制台输出 ###

> 线程运行结果45

## 是什么 ##

在我们需要并行运行程序时，可以集成Thread类和实现Runnable接口，**但是其没有返回值，也不能抛出异常**。因此添加Callable接口。  
但是**Thread只支持Runnable构造**，于是产生了FutureTask（**实现了runable以及Future接口**）支持使用Callable构造，那么FutureTask又是Runable可以交给Thread线程执行。所以FutureTask更像是个适配器，callable与Thread之间的适配器（类似将callble转为runable），当然FutureTask实现了Future接口，完成了其中获取任务运行状态，取消任务等功能。

## 继承结构 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70]

1.  Runnable  
    很熟悉了，内部只有一个run方法，并且是个函数式编程接口。在线程调用.start()方法后线程会调用其run方法
2.  Future  
    定义了一套关于人物的运行状态以及运行结果获取的一套接口

通过上述继承结构可以看出来FutureTask是一个Runable，那么Threa线程就可以通过FutureTask构造

## 核心属性 ##

1.  volatile int state; 任务的运行状态  
    static final int NEW = 0;  
    static final int COMPLETING = 1;  
    static final int NORMAL = 2;  
    static final int EXCEPTIONAL = 3;  
    static final int CANCELLED = 4;  
    static final int INTERRUPTING = 5;  
    static final int INTERRUPTED = 6;  
    状态的流转只有如下四条路径  
    NEW ->COMPLETING->NORMAL 任务正常执行完毕  
    NEW ->COMPLETING->EXCEPTIONAL 任务异常  
    NEW ->INTERRUPTING->INTERRUPTED 任务被中断  
    NEW -> ->CANCELLED 任务被取消
2.  Callable callable; 需要执行的任务，构造参数中传入
3.  Object outcome;输出，可能是异常的或者计算结果
4.  volatile Thread runner; 正在运行任务的线程
5.  WaitNode waiters; 一个等待队列的头节点（线程调用get()方法，如果任务没运行完，线程会阻塞挂起加入到等待队列）  
    static final class WaitNode \{  
    volatile Thread thread;  
    volatile WaitNode next;  
    WaitNode() \{ thread = Thread.currentThread(); \}  
    \}

## 运行原理图 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 1]

## 核心方法 ##

FutureTask 实现了Runable接口，那么一定就需要重写run()方法;线程Thread在调用.start()方法后，执行run方法（FutureTask不管是啥，终究是个Runable）。run方法中很明显的直接调用了callable的call方法。并且记录下call方法返回的返回值。

### public void run() ###

方法中调用了 Callable 的 call 方法获取其中的返回值，并且将返回值保存了起来。并不是直接return，因为future接口规定调用get方法获取返回值。在线程调用了get后再将暂存起来的值进行返回。

public void run() { 
      	// 如果状态不是 NEW，说明任务已经执行过了就不需要执行了直接返回，没有运行过则CAS把执行线程保存在runner字段
        // runnerOffset指的是runner字段的偏移量
        if (state != NEW ||!UNSAFE.compareAndSwapObject(this, runnerOffset,null, Thread.currentThread()))
            return;
        try { 
            Callable<V> c = callable;
            //CAS成功后还需要再检查一遍任务是否被运行
            if (c != null && state == NEW) { 
                V result;
                boolean ran;
                try { 
                  	// 正常调用 Callable 的 call 方法获取其中的返回值
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) { 
                    result = null;
                    ran = false;
                  	// catch Throwable catch所有的异常，统一保存起来，后续抛出
                    setException(ex);
                }
                // 运行成功，保存call方法的执行结果
                if (ran)
                    set(result);
            }
        } finally {         
            runner = null;       
            int s = state;
          	// 被中断了，进行中断处理
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }

### 保存结果（运算结果/异常结果） ###

CAS更新状态，并且将值赋值给outcome。最后唤醒被挂起的线程（文章开头说过调用get方法会将线程挂起，等待结果）。结果已经计算完成便可以唤醒所有阻塞线程

//保存异常
    protected void setException(Throwable t) { 
        //将状态从NEW->COMPLETING,更新状态为运行中
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { 
            outcome = t;
            //异常赋值完成，CAS状态为异常
            UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL);
            //唤醒调用了.get()阻塞被挂起的线程
            finishCompletion();
        }
    }
    // 保存运算结果
    protected void set(V v) { 
       //将状态从NEW->COMPLETING,更新状态为运行中
      if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { 
        outcome = v;
        //结果赋值完成，CAS状态为正常结束
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL);
        //唤醒调用了.get()在阻塞的线程
        finishCompletion();
      }
    }

### private void finishCompletion() ###

遍历waiters等待队列，将全部在等待结果的线程唤醒。

private void finishCompletion() { 
        //只要等待队列中有元素就要全部唤醒
        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
    }

### public V get() ###

需要获取任务的执行结果调用get()方法，但是这个结果时阻塞的。如果任务的执行状态为（normal，CANCELLED ，EXCEPTIONAL ，INTERRUPTED ）直接返回结果。否则调用awaitDone方法进一步判断线程是否需要挂起。

public V get() throws InterruptedException, ExecutionException { 
        int s = state;
      	// 如果 state 还没到 第三阶段（normal，CANCELLED ，EXCEPTIONAL ，INTERRUPTED ）
        // 说明outcome还不为结果的时候，则调用 awaitDone() 方法阻塞自己
        // 为COMPLETING 状态在awaitDone只会调用Thread.yield(); 让出cpu
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
      	// 返回结果
        return report(s);
    }

### private int awaitDone(boolean timed, long nanos) ###

线程判断自身是否需要被挂起。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 2]

private int awaitDone(boolean timed, long nanos) throws InterruptedException { 
      	// 计算等待截止时间
        final long deadline = timed ? System.nanoTime() + nanos : 0L;
        WaitNode q = null;
        boolean queued = false;
        for (;;) { 
          	// 如果当前线程被中断，如果是，则在等待队列中删除该节点，并抛出 InterruptedException
            if (Thread.interrupted()) { 
                removeWaiter(q);
                throw new InterruptedException();
            }
            int s = state;
          	// 状态大于 COMPLETING 说明已经达到最终状态（正常结束/异常结束/取消）
          	// 把 thread 置为空，并返回结果
            if (s > COMPLETING) { 
                if (q != null)
                    q.thread = null;
                return s;
            }
          	// 如果是COMPLETING 状态（中间状态），表示任务已结束，
            // 但 outcome 赋值还没结束，这时主动让出cpu执行权
            else if (s == COMPLETING) // cannot time out yet
                Thread.yield();
          	// 当前循环s<COMPLETING,将线程构造成节点，如果下次循环还是s<COMPLETING就入队
            else if (q == null)
              	// 将当前线程构造节点
                q = new WaitNode();
          	// 如果还没有入队列，则把当前节点加入waiters首节点并替换原来waiters
            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);
        }
    }

### public boolean cancel(boolean mayInterruptIfRunning) ###

取消任务

public boolean cancel(boolean mayInterruptIfRunning) { 
      
        if (!(state == NEW && UNSAFE.compareAndSwapInt(this, stateOffset, NEW,mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
            return false;
        try {   
          	// 需要中断任务执行线程
            if (mayInterruptIfRunning) { 
                try { 
                    Thread t = runner;
                  	// 中断线程
                    if (t != null)
                        t.interrupt();
                } finally {  
                  	// 修改为最终状态 INTERRUPTED
                    UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
                }
            }
        } finally { 
          	// 唤醒等待中的线程
            finishCompletion();
        }
        return true;
    }

[Link 1]: https://blog.csdn.net/wsdfym/article/details/110825953
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70]: /images/20221023/b23a33c67ee94b5d846a7079b1c1b699.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 1]: /images/20221023/24277d8b78fd41459667494625995d7e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 2]: /images/20221023/622ddf6b619b4a5ab432b383fc1244d1.png