并发编程（十六）：线程池

小鱼儿 2023-06-04 11:50 9阅读 0赞

# 一，ThreadPoolExecutor 概述 #

## 1，线程池优势 ##

在Java中，如果每个请求到达就创建一个线程，创建线程和销毁线程对系统资源的消耗都非常大，甚至可能比实际业务处理消耗的资源都大。同时，如果在JVM中创建太多的线程，也可能由于过度消耗内存或调度切换从而导致系统资源不足。

为了解决上面提出的问题，就有了线程池的概念。线程池，就是在一个线程容器中提前放置一定量的初始化线程，如果业务需要创建线程进行业务处理，则直接从线程池中获取一个线程进行执行，执行完成后归还线程到线程池，同时线程池也会对创建线程进行限制，不会无休止的创建下去。

**    使用线程池后，可以后下面几点优势：**

**        \* 减低创建线程和销毁线程的开销**

**        \* 提高响应速度，当有新任务需要执行时不需要等待线程创建时就可以直接执行（如果有空闲线程）**

**        \* 合理的设置线程池的大小可以避免因为硬件瓶颈带来的性能问题**

## 2，类图 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE5NzYzODg_size_16_color_FFFFFF_t_70][]

## 3，线程池常用API ##

// 线程池初始化
    public ThreadPoolExecutor(
    	// 核心线程数
    	int corePoolSize,
    	// 最大线程数
    	int maximumPoolSize,
    	// 线程空闲保留时间
    	long keepAliveTime,
    	// 线程保留单位
    	TimeUnit unit,
    	// 线程任务阻塞容器
    	BlockingQueue<Runnable> workQueue,
    	// 线程工厂，一般取默认
    	ThreadFactory threadFactory,
    	// 拒绝策略
    	RejectedExecutionHandler handler);
    	
    // 线程执行，不带返回值
    public void execute(Runnable command);
    
    // 线程执行，带返回值
    public <T> Future<T> submit(Callable<T> task);
    public <T> Future<T> submit(Runnable task, T result);
    public Future<?> submit(Runnable task);
    
    // 关闭线程池
    public void shutdown();

## 4，线程池常量解析，分为线程池常量和Future常量两部分 ##

/********************* 线程池常量 **********************/
    // 数量为，该值是29
    // 线程池把 Integer 的32位拆分为高3位和低29位，
    // 通过高三位表示状态，低29位表示正在执行的线程数量，
    private static final int COUNT_BITS = Integer.SIZE - 3;
    
    // 最大允许执行的线程数量，因为高三位表示状态，所以天然限制为29位
    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
    
    // 线程池运行状态
    // -1 的二进制是 11111111 11111111 11111111 11111111
    // 右移29位就是  11100000 00000000 00000000 00000000
    // 高三位表示状态，所以 RUNNING 的状态就是 111
    private static final int RUNNING    = -1 << COUNT_BITS;
    
    // 线程池关闭状态，不接收新任务，但是执行队列中的人物
    // 状态为 0
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    
    // 线程池停止状态，不接受新任务，不执行队列人物，终止执行任务
    // 状态为 1
    private static final int STOP       =  1 << COUNT_BITS;
    
    // 所有任务都已经结束，线程数量为0，处于该状态的线程池即将调用 terminated()方法
    // 状态为 10
    private static final int TIDYING    =  2 << COUNT_BITS;
    
    // terminated()方法执行完成
    // 状态为 11
    private static final int TERMINATED =  3 << COUNT_BITS;
    
    /********************* Future常量 **********************/
    // Future 常量主要是对 state 状态的几种情况分析
    // NEW 新建状态，表示这个 FutureTask还没有开始运行
    private static final int NEW = 0; 
    
    // COMPLETING 完成状态，表示 FutureTask 任务已经计算完毕了
    // 但是还有一些后续操作，例如唤醒等待线程操作，还没有完成。
    private static final int COMPLETING = 1;
    
    // FutureTask 任务完结，正常完成，没有发生异常
    private static final int NORMAL = 2;
    
    // FutureTask 任务完结，因为发生异常。
    private static final int EXCEPTIONAL = 3;
    
    // FutureTask 任务完结，因为取消任务
    private static final int CANCELLED = 4;
    
    // FutureTask 任务完结，也是取消任务，不过发起了中断运行任务线程的中断请求
    private static final int INTERRUPTING = 5;
    
    // FutureTask 任务完结，也是取消任务，已经完成了中断运行任务线程的中断请求
    private static final int INTERRUPTED = 6;

## 5，JDK提供的几种常用线程池，阿里开发手册不提倡用内置的线程池，建议通过构造器自行初始化，后续自行构造的执行流程会分析，该部分参考即可；跟源码可以发下，各种初始化方式最终也是通过构造器初始化！ ##

// 初始化定长线程池
    // 内置指定的核心线程数和最大线程数，并按照线程池流程执行
    public static ExecutorService newFixedThreadPool(int nThreads);
    
    // 初始化缓存的线程池
    // 没有核心线程数，默认最大线程数为 Integer.MAX_VALUE，阻塞队列为 SynchronousQueue，不保存数据
    // 所有对于缓存的线程池来说，接收一个任务的同时就需要执行一个任务
    public static ExecutorService newCachedThreadPool();
    
    // 初始化单例的线程池
    // 一次最多执行一个线程任务，多线程竞争时，添加到阻塞队列等候
    public static ExecutorService newSingleThreadExecutor();
    
    // 初始化定时的线程池
    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize);

## 6，功能DEMO ##

package com.gupao.concurrent;
    
    import java.util.concurrent.*;
    
    /**
     * @author pj_zhang
     * @create 2019-10-31 21:56
     **/
    public class ThreadPoolTest {
    
        private static ThreadPoolExecutor executor =
                new ThreadPoolExecutor(20, 20,
                        0L, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
    
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            Thread thread = new Thread(() -> {
                System.out.println("THREAD 线程执行");
            });
    
            Runnable runnable = () -> {
                System.out.println("RUNNABLE 线程执行");
            };
    
            Callable<String> callable = () -> {
                Thread.sleep(2000);
                return "Callable 线程执行";
            };
    
            executor.execute(thread);
            executor.execute(runnable);
            // 此处打印时间，是为了演示 Future 的阻塞获取结果
            System.out.println("callable执行前：" + System.currentTimeMillis());
            Future<String> future = executor.submit(callable);
            String callableResult = future.get();
            System.out.println("callable执行后：" + System.currentTimeMillis() + ", " + callableResult);
        }
    
    }

![2019103122063457.png][]

# 二，源码分析 #

## 1，底层方法分析 ##

### 1.1，ctlOf(int rs, int wc)：获取线程池状态+数量的 Integer 值， ###

private static int ctlOf(int rs, int wc) { 
    	return rs | wc; 
    }

### 1.2，workerCountOf(int c)：获取工作线程数量 ###

private static int workerCountOf(int c)  {
    	// 根据 ctlOf 获取到的值，用低29位进行与运算，获取到线程数量
    	return c & CAPACITY; 
    }

### 1.3，runStateOf(int c)：获取线程状态 ###

private static int runStateOf(int c)     { 
        return c & ~CAPACITY; 
    }

## 2，ThreadPoolExecutor 初始化及执行流程 ##

### 2.1，ThreadPoolExecutor() ###

public ThreadPoolExecutor(int corePoolSize,
    						  int maximumPoolSize,
    						  long keepAliveTime,
    						  TimeUnit unit,
    						  BlockingQueue<Runnable> workQueue,
    						  ThreadFactory threadFactory,
    						  RejectedExecutionHandler handler) {
    	if (corePoolSize < 0 ||
    		maximumPoolSize <= 0 ||
    		maximumPoolSize < corePoolSize ||
    		keepAliveTime < 0)
    		throw new IllegalArgumentException();
    	if (workQueue == null || threadFactory == null || handler == null)
    		throw new NullPointerException();
    	this.acc = System.getSecurityManager() == null ?
    			null :
    			AccessController.getContext();
    	// 核心线程数
    	this.corePoolSize = corePoolSize;
    	// 最大线程数
    	this.maximumPoolSize = maximumPoolSize;
    	// 阻塞队列
    	this.workQueue = workQueue;
    	// 保持存活时间
    	this.keepAliveTime = unit.toNanos(keepAliveTime);
    	// 线程工厂，这个直接取默认
    	this.threadFactory = threadFactory;
    	// 拒绝策略
    	this.handler = handler;
    }

### 2.2，线程池执行流程 ###

\* 接收到线程任务后，首先判断核心线程有没有被全部占用；没有被全部占用，随机构建一个线程执行任务

\* 如果核心线程全部被占用，查看阻塞队列是否已满；如果没有满，添加到阻塞队列

\* 如果阻塞队列已满，继续看最大线程数有没有被全部占用；如果存在空闲，构建线程执行任务

\* 如果最大线程数已经全部占用，根据定义的拒绝策略进行拒绝操作

### 2.3，线程池拒绝策略 ###

\* 线程池提供了四种拒绝策略，分别是 *RejectedExecutionHandler* 接口的四种实现

// java.util.concurrent.ThreadPoolExecutor.AbortPolicy
    // 异常处理
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    	throw new RejectedExecutionException("Task " + r.toString() +
    										 " rejected from " +
    										 e.toString());
    }
    
    // java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy
    // 丢弃最前面的任务，重新添加执行
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    	if (!e.isShutdown()) {
    		e.getQueue().poll();
    		e.execute(r);
    	}
    }
    
    // java.util.concurrent.ThreadPoolExecutor.DiscardPolicy
    // 什么都不做，即丢弃当前任务
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    }
    
    // java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy
    // 只要线程池没有关闭，则直接开线程运行，该策略建议慎用，不收控制
    public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
    	if (!e.isShutdown()) {
    		r.run();
    	}
    }
    
    // 最后，业务可以自定义拒绝方式，只需要实现 RejectedExecutionHandler 接口，然后重写其接口方法

## 3，execute() ##

\* execute(Runnable command)：刚才对线程池的大概执行流程进行了分析，该方法内可以看出代码实现

public void execute(Runnable command) {
    	if (command == null)
    		throw new NullPointerException();
    	// 获取状态位 + 数量位的 int 对象
    	int c = ctl.get();
    	// 获取工作线程数量，首先判断核心线程数
    	if (workerCountOf(c) < corePoolSize) {
    		// 存在空闲的核心线程，进行线程执行
    		if (addWorker(command, true))
    			return;
    		// 如果执行失败，重新对 c 赋值，说明存在线程竞争
    		c = ctl.get();
    	}
    	// isRunning(c)：线程池依旧运行状态
    	// workQueue.offer(command)：添加到队列成功
    	// 次数是判断加入到阻塞队列是否成功
    	if (isRunning(c) && workQueue.offer(command)) {
    		int recheck = ctl.get();
    		// 如果线程池不处于运行状态，则从队列中移除当前任务
    		// 并执行拒绝策略
    		if (! isRunning(recheck) && remove(command))
    			reject(command);
    		// 如果之前的线程已经被销毁完，新建一个线程
    		else if (workerCountOf(recheck) == 0)
    			addWorker(null, false);
    	}
    	// 阻塞队列添加失败，唤醒最大线程执行，
    	else if (!addWorker(command, false))
    		// 最大线程执行失败，进行拒绝策略处理
    		reject(command);
    }

\* addWorker(Runnable firstTask, boolean core)

// Runnable firstTask：当前线程任务
    // boolean core：是否核心线程
    private boolean addWorker(Runnable firstTask, boolean core) {
    	retry:
    	for (;;) {
    		// 获取线程执行状态
    		int c = ctl.get();
    		// 这部分计算后续搞明白再填充 TODO
    		int rs = runStateOf(c);
    		// 线程池已经关闭，不再接受新任务
    		// SHUTDOWN 状态不接受新任务，但仍然会执行已经加入任务队列的任务
    		// 所以当进入 SHUTDOWN 状态，而传进来的任务为空，并且任务队列不为空的时候，是允许添加新线程的,
    		// 如果把这个条件取反，就表示不允许添加 worker
    		if (rs >= SHUTDOWN &&
    			! (rs == SHUTDOWN &&
    			   firstTask == null &&
    			   ! workQueue.isEmpty()))
    			return false;
    		// 自旋添加任务
    		for (;;) {
    			// 获取工作线程
    			int wc = workerCountOf(c);
    			// 首先判断是否大于最大允许数量
    			// 然后根据是否核心线程判断是否大于核心线程数或者最大线程数
    			if (wc >= CAPACITY ||
    				wc >= (core ? corePoolSize : maximumPoolSize))
    				return false;
    			// 对c递增，也就是工作线程数递增
    			if (compareAndIncrementWorkerCount(c))
    				break retry;
    			// 递增失败，说明存在线程竞争或者状态变更，继续自旋处理
    			c = ctl.get(); 
    			// 此处不等于，说明存在线程池状态变更
    			// 等于，说明只是存在线程竞争造成的CAS失败
    			if (runStateOf(c) != rs)
    				continue retry;
    		}
    	}
    
    	// 上半部分基本是对线程池状态及工作线程数进行判断，并最终对工作线程+1，表示当前线程已经抢占到一个线程位置
    	boolean workerStarted = false;
    	boolean workerAdded = false;
    	Worker w = null;
    	try {
    		// 包装线程对象为 Worker 对象
    		w = new Worker(firstTask);
    		// 通过 ThreadFactory 构建一个新的线程
    		final Thread t = w.thread;
    		if (t != null) {
    			// 此处加重入锁
    			final ReentrantLock mainLock = this.mainLock;
    			mainLock.lock();
    			try {
    				// 继续获取线程池状态进行判断
    				int rs = runStateOf(ctl.get());
    
    				if (rs < SHUTDOWN ||
    					(rs == SHUTDOWN && firstTask == null)) {
    					// 如果线程已经运行中，则说明存在问题，此处线程还没有启用
    					if (t.isAlive())
    						throw new IllegalThreadStateException();
    					// 添加线程包装后的Worker对象到列表中
    					workers.add(w);
    					int s = workers.size();
    					if (s > largestPoolSize)
    						largestPoolSize = s;
    					// 表示工作线程创建成功
    					workerAdded = true;
    				}
    			} finally {
    				mainLock.unlock();
    			}
    			if (workerAdded) {
    				// 线程启动，
    				// 此处注意，Worker 实现了 Runnable接口，则此处是调用 Worker.run()
    				t.start();
    				workerStarted = true;
    			}
    		}
    	} finally {
    		// 如果添加失败，则递减工作线程数
    		if (! workerStarted)
    			addWorkerFailed(w);
    	}
    	return workerStarted;
    }

\* runWorker(Worker w)：Worker的 run() 方法内部调用该方法

final void runWorker(Worker w) {
    	Thread wt = Thread.currentThread();
    	// 获取初始化 Worker时传递的线程
    	// 此处分析的是 execute() 触发，如果是 submit() 触发，此处的Runnable实现类应该为 FutureTask，
    	// task.run() 最终调用 FutureTask.run()方法，此处会对 Future的状态进行处理，实现阻塞获取的功能
    	Runnable task = w.firstTask;
    	w.firstTask = null;
    	// unlock，表示当前 worker 线程允许中断，因为 new Worker 默认的 state=-1,
    	// 此处是调用Worker 类的 tryRelease()方法，将 state 置为 0， 
    	// 而 interruptIfStarted()中只有 state>=0 才允许调用中断
    	w.unlock();
    	boolean completedAbruptly = true;
    	try {
    		// 任务不为空，则持续执行
    		// getTask()：此处表示不断从阻塞队列中获取元素
    		while (task != null || (task = getTask()) != null) {
    			w.lock();
    			// 线程池状态为stop时不接受新任务，并中断正在执行的人物
    			// (Thread.interrupted() &&runStateAtLeast(ctl.get(), STOP)确保线程中断标志位为 true 且是 stop 状态以上，接着清除了中断标志
    			// !wt.isInterrupted()则再一次检查保证线程需要设置中断标志位
    			if ((runStateAtLeast(ctl.get(), STOP) ||
    				 (Thread.interrupted() &&
    				  runStateAtLeast(ctl.get(), STOP))) &&
    				!wt.isInterrupted())
    				wt.interrupt();
    			try {
    				// 默认没有实现功能，前置处理
    				beforeExecute(wt, task);
    				Throwable thrown = null;
    				try {
    					// 此处注意 task.run() 的task的不同实现
    					// 在submit()触发的功能中，表示FutureTask
    					task.run();
    				} catch (RuntimeException x) {
    					thrown = x; throw x;
    				} catch (Error x) {
    					thrown = x; throw x;
    				} catch (Throwable x) {
    					thrown = x; throw new Error(x);
    				} finally {
    					// 后置执行，没有处理
    					afterExecute(task, thrown);
    				}
    			} finally {
    				task = null;
    				w.completedTasks++;
    				w.unlock();
    			}
    		}
    		completedAbruptly = false;
    	} finally {
    		// 将入参 worker 从数组 workers 里删除掉；
    		// 根据布尔值 allowCoreThreadTimeOut 来决定是否补充新的 Worker 进数组workers
    		processWorkerExit(w, completedAbruptly);
    	}
    }

\* getTask()：从阻塞队列中获取下一个有效任务。线程池定义的超时处理再该部分实现

private Runnable getTask() {
    	boolean timedOut = false;
    
    	for (;;) {
    		int c = ctl.get();
    		int rs = runStateOf(c);
    
    		// 校验线程池状态
    		if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
    			decrementWorkerCount();
    			return null;
    		}
    
    		// 获取工作线程
    		int wc = workerCountOf(c);
    		// 对超时线程进行时间控制
    		// allowCoreThreadTimeOut默认为false，表示核心线程不收控制
    		// wc > corePoolSize：超过核心线程，即最大线程数，则触发控制
    		boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    
    		// timedOut为true，说明上次阻塞操作已经超时，则工作线程数-1，并返回null
    		if ((wc > maximumPoolSize || (timed && timedOut))
    			&& (wc > 1 || workQueue.isEmpty())) {
    			if (compareAndDecrementWorkerCount(c))
    				return null;
    			continue;
    		}
    
    		try {
    			// 此处就是对超时控制的处理，在从队列中获取数据时，阻塞获取
    			Runnable r = timed ?
    				workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
    				workQueue.take();
    			// 如果拿到任务，则直接返回去进行处理
    			if (r != null)
    				return r;
    			// 走到这一步，说明超时，在下一步时候进行回收处理
    			timedOut = true;
    		} catch (InterruptedException retry) {
    			timedOut = false;
    		}
    	}
    }

\* addWorkerFailed(Worker w)：添加工作线程失败

private void addWorkerFailed(Worker w) {
    	final ReentrantLock mainLock = this.mainLock;
    	mainLock.lock();
    	try {
    		// 从列表中移除当前 Worker
    		if (w != null)
    			workers.remove(w);
    		// 工作线程数递减
    		decrementWorkerCount();
    		// 尝试修改线程状态为 Terminate
    		tryTerminate();
    	} finally {
    		mainLock.unlock();
    	}
    }
    
    private void decrementWorkerCount() {
    	do {} while (! compareAndDecrementWorkerCount(ctl.get()));
    }

\* reject(Runnable command)：拒绝策略

final void reject(Runnable command) {
    	// 直接执行拒绝策略
    	handler.rejectedExecution(command, this);
    }

## 4，submit()：同时对 Future 进行分析 ##

\* submit(Callable<T> task)：触发线程执行

public <T> Future<T> submit(Callable<T> task) {
    	if (task == null) throw new NullPointerException();
    	// 初始化化一个 FutureTask，实现了 Runnable 接口
    	RunnableFuture<T> ftask = newTaskFor(task);
    	// 直接执行 execute 方法
    	execute(ftask);
    	return ftask;
    }
    
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    	return new FutureTask<T>(callable);
    }

\* 在 runWorker() 方法中，触发 task.run()，实际调用的是 Future.run()方法

public void run() {
    	// 状态不为new，直接执行异常
    	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 {
    				// 直接调用Callable的call方法，并返回结果
    				result = c.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);
    	}
    }

\* set(V v)：设置执行结果

protected void set(V v) {
    	// 设置状态为完成
    	if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
    		outcome = v;
    		// 设置正常结束
    		UNSAFE.putOrderedInt(this, stateOffset, NORMAL);
    		// 完成后续处理，唤醒等待节点
    		finishCompletion();
    	}
    }

\* finishCompletion()：该方法主要是唤醒等待节点，去返回结果中拿数据

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
    }

## 5，Funture.get() ##

\* get()：阻塞获取数据

public V get() throws InterruptedException, ExecutionException {
    	// 获取Future对应的线程执行状态，如果没有执行完直接等待
    	int s = state;
    	if (s <= COMPLETING)
    		s = awaitDone(false, 0L);
    	// 执行完成后，解析结果集
    	return report(s);
    }

\* awaitDone(boolean timed, long nanos)

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();
    		// 表示状态为null，构建等待节点，准备等待
    		else if (q == null)
    			q = new WaitNode();
    		// 使用CAS添加等待节点到等待队列
    		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);
    	}
    }

\* report(int s)：获取返回值

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

# #

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTE5NzYzODg_size_16_color_FFFFFF_t_70]: /images/20230601/273d51dac21c45179d702acf87c3e748.png
[2019103122063457.png]: /images/20230601/49f1699363a94f99a6fcc2d053aa5dad.png