九. Executor框架中的任务和任务的执行结果

你的名字 2024-03-29 14:43 0阅读 0赞

### 前言 ###

`Executor`框架中，提供了`Runnable`接口和`Callable`接口用于定义任务，提供了`Future`接口来表示任务的**异步计算结果**，本篇文章将对任务和任务的执行结果进行分析。

### 正文 ###

#### 一. 任务 ####

在`Executor`框架中，`Runnable`接口和`Callable`接口用于定义任务，`Runnable`接口的实现类可以被`Thread`，`ThreadPoolExecutor`和`ScheduledThreadPoolExecutor`执行，`Callable`接口的实现类可以被`ThreadPoolExecutor`和`ScheduledThreadPoolExecutor`执行，此外，`Runnable`接口和`Callable`接口最大的不同在于：**`Runnable`任务没有返回值，`Callable`任务有返回值。**

`Executor`框架提供的工具类`Executors`可以将`Runnable`封装成`Callable`，方法签名如下所示。

public static Callable<Object> callable(Runnable task)
    
    public static <T> Callable<T> callable(Runnable task, T result)

#### 二. 任务的执行结果 ####

`Executor`框架提供了`Future`接口来表示任务的**异步计算结果**。`Future`接口定义如下所示。

public interface Future<V> {
        
    
    	// 关闭任务的执行，在任务已经执行完毕或者已经被关闭时，返回false
    	// 在该方法调用时如果任务正在被执行，mayInterruptIfRunning决定是否打断执行任务的线程来停止任务
        boolean cancel(boolean mayInterruptIfRunning);
    
    	// 判断任务是否被关闭过
    	// 如果在任务执行完之前执行过关闭任务的操作，该方法返回true
        boolean isCancelled();
    
        // 判断任务是否执行完毕
    	// 因为中断，异常和关闭而导致的任务执行完毕，该方法均会返回true
        boolean isDone();
    
        // 等待任务执行完毕并获取执行结果
        V get() throws InterruptedException, ExecutionException;
    
        // 在指定时间内等待任务执行完毕并获取执行结果
        V get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException;
    }

此外，`Executor`框架中还有一个`RunnableFuture`接口，该接口继承于`Runnable`接口和`Future`接口，**即该接口的实现类即可以作为`Runnable`被执行，也能作为`Future`获取执行结果。**

所以`Executor`框架提供了一个`RunnableFuture`接口的实现类`FutureTask`，而且，在调用`ThreadPoolExecutor`的`submit()`方法提交任务（`Runnable`或者`Callable`）或者向`ScheduledThreadPoolExecutor`提交任务（`Runnable`或者`Callable`）时，所提交的任务均会被封装为`FutureTask`，然后封装成`FutureTask`的任务会作为`Runnable`被添加到任务阻塞队列中，同时也会作为`Future`被返回。

`FutureTask`的类图如下所示。

![在这里插入图片描述][67b74f3ccd314cf4a3595763c8391e85.jpeg_pic_center]

#### 三. FutureTask ####

`FutureTask`是`Executor`框架中重要的组件，下面对其原理进行学习。`FutureTask`的关键字段如下所示。

public class FutureTask<V> implements RunnableFuture<V> {
        
    
        // 任务状态，会有以下几种变化情况：
    	// NEW -> COMPLETING -> NORMAL
    	// NEW -> COMPLETING -> EXCEPTIONAL
    	// NEW -> CANCELLED
    	// NEW -> INTERRUPTING -> INTERRUPTED
        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;
    
        // 需要被执行的任务
        private Callable<V> callable;
        // 任务执行的结果或抛出的异常
        private Object outcome;
        // 正在执行任务的线程
        private volatile Thread runner;
        // 记录调用get()方法等待的线程
        private volatile WaitNode waiters;
    	
    	......
    
    }

关于`FutureTask`的状态字段稍后再分析，现在先看一下`FutureTask`中的需要被执行的任务**callable**字段，可知该字段为`Callable`接口，而前面已经分析知道无论是`ThreadPoolExecutor`还是`ScheduledThreadPoolExecutor`，均可以将`Runnable`或`Callable`封装为`FutureTask`，而`FutureTask`中却只有`Callable`，那么肯定是在某个地方将`Runnable`转换为了`Callable`，这个地方就是`FutureTask`的构造函数，如下所示。

public FutureTask(Callable<V> callable) {
        
    	if (callable == null)
    		throw new NullPointerException();
    	this.callable = callable;
    	this.state = NEW;
    }
    
    public FutureTask(Runnable runnable, V result) {
        
    	this.callable = Executors.callable(runnable, result);
    	this.state = NEW;
    }

可以看到，`FutureTask`的构造函数中，如果传进来的任务为`Runnable`，那么会使用工具类`Executors`将`Runnable`转换为`Callable`。同时构造函数中还会将状态设置为**NEW**。

下面分析`FutureTask`的`run()`方法，源码如下所示。

public void run() {
        
    	// 判断任务状态是否为NEW，状态为NEW的任务才允许被执行
    	// 如果任务状态为NEW，则以CAS方式将任务的runner字段设置为当前执行任务的线程，设置成功才允许执行任务
    	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();
    				ran = true;
    			} catch (Throwable ex) {
        
    				result = null;
    				ran = false;
    				// 任务执行时抛出异常，将异常赋值给outcome字段
    				// 然后以CAS方式将状态先置为COMPLETING，然后置为EXCEPTIONAL，最后唤醒所有调用get()方法进入等待的线程
    				setException(ex);
    			}
    			if (ran)
    				// 任务执行时未抛出异常，将执行结果赋值给outcome字段
    				// 然后以CAS方式将状态先置为COMPLETING，然后置为NORMAL，最后唤醒所有调用get()方法进入等待的线程
    				set(result);
    		}
    	} finally {
        
    		// 在任务执行结束后将runner字段置为null
    		// 在任务执行结束以前runner字段不能为null，以防止任务被并发多次执行
    		runner = null;
    		int s = state;
    		if (s >= INTERRUPTING)
    			// 执行到这里，表示任务在执行时（任务状态为NEW）被调用了cancel(true)方法，并且cancel(true)中将任务状态置为了INTERRUPTING或INTERRUPTED
    			// 如果任务状态为INTERRUPTING，则循环调用Thread.yield()来放弃时间片，直到任务状态变为INTERRUPTED
    			handlePossibleCancellationInterrupt(s);
    	}
    }

在`run()`方法中，主要的步骤如下。

*  先在任务状态为**NEW**的情况下以**CAS**方式将执行任务的线程**runner**字段置为当前线程，任务状态不为**NEW**或者**CAS**失败，都直接退出`run()`方法，防止任务被并发多次执行；
 *  任务执行成功或者任务执行抛出异常，会分别以**CAS**方式将任务状态先置为**COMPLETING**，如果**CAS**成功，之后调用`cancel()`方法会直接返回**false**，但是如果**CAS**操作之前先调用了`cancel()`方法，那么会导致**CAS**失败，此时任务的状态为**INTERRUPTING**，**CANCELLED**或者**INTERRUPTED**；
 *  在`run()`方法最后会判断任务的状态是否为**INTERRUPTING**或**INTERRUPTED**，如果满足，表明在任务执行完毕并翻转任务状态之前`cancel(true)`方法被调用了，此时如果任务状态为**INTERRUPTING**，则循环调用`Thread.yield()`来放弃时间片以等待任务状态翻转为**INTERRUPTED**。

同时`FutureTask`还提供了一个`runAndReset()`方法，该方法与`run()`方法略有不同，如下所示。

*  `runAndReset()`不关心任务的执行结果，即不会将任务执行结果赋值给**outcome**字段；
 *  `runAndReset()`不会主动去翻转任务的状态，即任务正常执行完毕之后状态还是为**NEW**，以适用于任务需要多次被执行的情况，比如定时任务。

下面分析`FutureTask`的`get()`方法，`FutureTask`提供了一直等待直到任务执行完毕的`get()`方法，和指定等待时间的`get()`方法，如下所示。

public V get() throws InterruptedException, ExecutionException {
        
    	int s = state;
    	if (s <= COMPLETING)
    		s = awaitDone(false, 0L);
    	return report(s);
    }
    
    public V get(long timeout, TimeUnit unit)
    	throws InterruptedException, ExecutionException, TimeoutException {
        
    	if (unit == null)
    		throw new NullPointerException();
    	int s = state;
    	if (s <= COMPLETING &&
    		(s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
    		throw new TimeoutException();
    	return report(s);
    }

在`get()`方法中，会先在`awaitDone()`方法中获取任务状态，如果任务状态为**NEW**，则进入等待状态。获取到任务状态之后，在`report()`方法中根据获取到的任务状态来返回任务执行结果。

`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()) {
        
    			// 调用get()方法的线程如果被中断，则将其从等待链表中删除，并抛出中断异常
    			removeWaiter(q);
    			throw new InterruptedException();
    		}
    
    		int s = state;
    		if (s > COMPLETING) {
        
    			// 如果任务状态大于COMPLETING，说明任务执行完毕，或抛出异常，或被调用了cancel()方法
    			if (q != null)
    				q.thread = null;
    			// 返回任务的状态
    			return s;
    		}
    		else if (s == COMPLETING)
    			// 如果任务状态为COMPLETING，则放弃时间片，目的是为了调用get()方法的线程再次获得时间片时任务状态翻转为NORMAL或者EXCEPTIONAL
    			Thread.yield();
    		else if (q == null)
    			// 执行到这里，说明任务状态为NEW
    			// 基于调用get()方法的线程创建一个WaitNode节点
    			q = new WaitNode();
    		else if (!queued)
    			// 如果WaitNode节点没有添加到等待链表，则将其加入到等待链表中
    			queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
    												 q.next = waiters, q);
    		else if (timed) {
        
    			// 如果调用get()方法时指定了等待时间，则使用LockSupport进入等待状态并指定等待时间
    			// 如果等待时间到，任务状态还是NEW，则移除WaitNode节点并返回任务状态
    			nanos = deadline - System.nanoTime();
    			if (nanos <= 0L) {
        
    				removeWaiter(q);
    				return state;
    			}
    			LockSupport.parkNanos(this, nanos);
    		}
    		else
    			// 调用get()方法的线程进入等待状态
    			LockSupport.park(this);
    	}
    }

`awaitDone()`中，主要步骤如下所示。

*  如果任务状态大于**COMPLETING**，即任务执行完毕，或抛出异常，或被调用了`cancel()`方法，此时返回任务状态；
 *  如果任务状态为**COMPLETING**，表示任务已经执行完毕（或抛出异常），但是执行结果尚未赋值给**outcome**字段，此时调用`Thread.yield()`放弃时间片，因为任务状态从**COMPLETING**向**NORMAL**或**EXCEPTIONAL**转换的时间非常短，所以`Thread.yield()`能够让调用`get()`方法的线程更快的响应任务状态的转换，最终目的是为了让调用`get()`方法的线程再次获得时间片时任务状态已经翻转为**NORMAL**或者**EXCEPTIONAL**；
 *  如果任务状态为**NEW**，那么说明任务还未开始执行或者任务正在执行，此时基于调用`get()`方法的线程创建一个`WaitNode`节点并加入等待链表中；
 *  调用`get()`方法的线程进入等待状态，只有在等待时间到（如果指定了的话），或者任务执行完毕，或者任务执行抛出异常，或者`cancel()`方法被调用时，所有等待线程才会结束等待状态。

`report()`方法的实现如下所示。

private V report(int s) throws ExecutionException {
        
    	Object x = outcome;
    	if (s == NORMAL)
    		return (V)x;
    	if (s >= CANCELLED)
    		throw new CancellationException();
    	throw new ExecutionException((Throwable)x);
    }

下面最后看一下`cancel()`方法的实现，如下所示。

public boolean cancel(boolean mayInterruptIfRunning) {
        
    	// 如果任务状态不是NEW，则返回false
    	// 如果以CAS方式将任务状态置为INTERRUPTING或者CANCELLED失败，则返回false
    	// 即任务已经执行完毕或者已经被关闭时，返回false
    	if (!(state == NEW &&
    		  UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
    			  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
    		return false;
    	try {
        
    		// mayInterruptIfRunning为true表示需要中断正在执行任务的线程，并最终将任务状态置为INTERRUPTED
    		if (mayInterruptIfRunning) {
        
    			try {
        
    				Thread t = runner;
    				if (t != null)
    					t.interrupt();
    			} finally {
        
    				UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
    			}
    		}
    	} finally {
        
    		// 唤醒所有调用get()方法进入等待的线程
    		finishCompletion();
    	}
    	return true;
    }

`cancel()`方法调用时如果任务已经执行完毕（状态为**COMPLETING**，**NORMAL**或**EXCEPTIONAL**），或者任务已经被关闭（状态为**CANCELLED**，**INTERRUPTING**或**INTERRUPTED**），则直接返回**false**，并且会根据**mayInterruptIfRunning**参数来决定是否会中断正在执行任务的线程。

### 总结 ###

`Executor`框架中，`Runnable`接口和`Callable`接口表示需要被执行的任务，不同在于前者无返回值而后者有返回值。`Future`接口表示异步计算的结果，同时`RunnableFuture`接口继承了`Runnable`接口和`Future`接口，`RunnableFuture`接口的实现类`FutureTask`既能作为`Runnable`任务来执行，也能作为`Future`来获取计算结果。

对于`RunnableFuture`接口的实现类`FutureTask`，可以通过调用`FutureTask`的`cancel(true)`方法来中断正在执行任务的线程，但是请注意，如果线程执行的任务不响应中断，那么这个任务是不会被停止的。

[67b74f3ccd314cf4a3595763c8391e85.jpeg_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/865e7d7741e4479cb8e6d39ab33b7dd3.png