简单研究下线程池的执行原理。以及execute 和 submit 方法的区别。
1. execute 方法接收的是一个Runnable 参数，返回值是void 类型，也就是不接收结果， 方法签名如下：
java.util.concurrent.Executor#execute
void execute(Runnable command);
2. submit 方法三个重载的方法，如下：
对应的方法如下：‘
(1) java.util.concurrent.ExecutorService#submit(java.lang.Runnable)
Future<?> submit(Runnable task);
(2) java.util.concurrent.ExecutorService#submit(java.lang.Runnable, T)
<T> Future<T> submit(Runnable task, T result);
(3) java.util.concurrent.ExecutorService#submit(java.util.concurrent.Callable<T>)
<T> Future<T> submit(Callable<T> task);
1.java.util.concurrent.ThreadPoolExecutor#execute理解
测试代码
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(3)); // execute 方法无返回值，相当于执行不带返回结果的任务 threadPoolExecutor.execute(() -> { log.info("--" + Thread.currentThread().getName() + "--A"); try { Thread.sleep(2 * 1000); } catch (InterruptedException e) { } log.info("--" + Thread.currentThread().getName() + "--B"); });
查看源代码如下：
(1) 接口是Executor 的方法
void execute(Runnable command);
(2) 实现类java.util.concurrent.ThreadPoolExecutor#execute
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); /* * Proceed in 3 steps: * * 1. If fewer than corePoolSize threads are running, try to * start a new thread with the given command as its first * task. The call to addWorker atomically checks runState and * workerCount, and so prevents false alarms that would add * threads when it shouldn't, by returning false. * * 2. If a task can be successfully queued, then we still need * to double-check whether we should have added a thread * (because existing ones died since last checking) or that * the pool shut down since entry into this method. So we * recheck state and if necessary roll back the enqueuing if * stopped, or start a new thread if there are none. * * 3. If we cannot queue task, then we try to add a new * thread. If it fails, we know we are shut down or saturated * and so reject the task. */ int c = ctl.get(); if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } 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); }
源码跟踪：
(1) addWorker(command, true) 增加worker
private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); // Check if queue empty only if necessary. 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; if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); // Re-read ctl if (runStateOf(c) != rs) continue retry; // else CAS failed due to workerCount change; retry inner loop } } boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // Recheck while holding lock. // Back out on ThreadFactory failure or if // shut down before lock acquired. int rs = runStateOf(ctl.get()); if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) { if (t.isAlive()) // precheck that t is startable throw new IllegalThreadStateException(); workers.add(w); int s = workers.size(); if (s > largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { t.start(); workerStarted = true; } } } finally { if (! workerStarted) addWorkerFailed(w); } return workerStarted; }
worker 是一个内部类java.util.concurrent.ThreadPoolExecutor.Worker，其包含的重要属性和方法如下：
private final class Worker extends AbstractQueuedSynchronizer implements Runnable { /** * This class will never be serialized, but we provide a * serialVersionUID to suppress a javac warning. */ private static final long serialVersionUID = 6138294804551838833L; /** Thread this worker is running in. Null if factory fails. */ final Thread thread; /** Initial task to run. Possibly null. */ Runnable firstTask; /** Per-thread task counter */ volatile long completedTasks; Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } public void run() { runWorker(this); } final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { while (task != null || (task = getTask()) != null) { w.lock(); // If pool is stopping, ensure thread is interrupted; // if not, ensure thread is not interrupted. This // requires a recheck in second case to deal with // shutdownNow race while clearing interrupt 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(); } 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 { processWorkerExit(w, completedAbruptly); } } ...
1》new Worker 的时候会创建线程，线程传入的Runnable 对象是Worker 自身，同时Worker
2》t.start(); 启动的时候会调用java.util.concurrent.ThreadPoolExecutor.Worker#run 方法
3》然后调用到java.util.concurrent.ThreadPoolExecutor#runWorker， 方法内部调用task.run(); 相当于同步的调用 传给线程池的Runnable 任务的run 方法。
4》当不是第一个任务的时候走的是java.util.concurrent.ThreadPoolExecutor#getTask， 也就是从任务队列中阻塞拿去任务
private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? for (;;) { int c = ctl.get(); int rs = runStateOf(c); // Check if queue empty only if necessary. if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); // Are workers subject to culling? boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; 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; } } }
2. submit理解
1. java.util.concurrent.AbstractExecutorService#submit(java.lang.Runnable) 理解
查看源码如下：java.util.concurrent.AbstractExecutorService#submit(java.lang.Runnable)
public Future<?> submit(Runnable task) { if (task == null) throw new NullPointerException(); RunnableFuture<Void> ftask = newTaskFor(task, null); execute(ftask); return ftask; }
可以理解大体的思路是对，传给线程池的Runnable 进行了下包装，java.util.concurrent.AbstractExecutorService#newTaskFor(java.lang.Runnable, T) 如下：
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { return new FutureTask<T>(runnable, value); }
(1) java.util.concurrent.RunnableFuture 继承了接口Runnable, Future<V>
public interface RunnableFuture<V> extends Runnable, Future<V> { void run(); }
java.util.concurrent.Future 接口如下：
public interface Future<V> { /** * Attempts to cancel execution of this task. This attempt will * fail if the task has already completed, has already been cancelled, * or could not be cancelled for some other reason. If successful, * and this task has not started when {@code cancel} is called, * this task should never run. If the task has already started, * then the {@code mayInterruptIfRunning} parameter determines * whether the thread executing this task should be interrupted in * an attempt to stop the task. * * <p>After this method returns, subsequent calls to {@link #isDone} will * always return {@code true}. Subsequent calls to {@link #isCancelled} * will always return {@code true} if this method returned {@code true}. * * @param mayInterruptIfRunning {@code true} if the thread executing this * task should be interrupted; otherwise, in-progress tasks are allowed * to complete * @return {@code false} if the task could not be cancelled, * typically because it has already completed normally; * {@code true} otherwise */ boolean cancel(boolean mayInterruptIfRunning); /** * Returns {@code true} if this task was cancelled before it completed * normally. * * @return {@code true} if this task was cancelled before it completed */ boolean isCancelled(); /** * Returns {@code true} if this task completed. * * Completion may be due to normal termination, an exception, or * cancellation -- in all of these cases, this method will return * {@code true}. * * @return {@code true} if this task completed */ boolean isDone(); /** * Waits if necessary for the computation to complete, and then * retrieves its result. * * @return the computed result * @throws CancellationException if the computation was cancelled * @throws ExecutionException if the computation threw an * exception * @throws InterruptedException if the current thread was interrupted * while waiting */ V get() throws InterruptedException, ExecutionException; /** * Waits if necessary for at most the given time for the computation * to complete, and then retrieves its result, if available. * * @param timeout the maximum time to wait * @param unit the time unit of the timeout argument * @return the computed result * @throws CancellationException if the computation was cancelled * @throws ExecutionException if the computation threw an * exception * @throws InterruptedException if the current thread was interrupted * while waiting * @throws TimeoutException if the wait timed out */ V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException; }
java.util.concurrent.FutureTask 的重要属性和方法如下：
public class FutureTask<V> implements RunnableFuture<V> { 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; /** The underlying callable; nulled out after running */ private Callable<V> callable; /** The result to return or exception to throw from get() */ private Object outcome; // non-volatile, protected by state reads/writes /** The thread running the callable; CASed during run() */ private volatile Thread runner; /** Treiber stack of waiting threads */ private volatile WaitNode waiters; 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 } public boolean isCancelled() { return state >= CANCELLED; } public boolean isDone() { return state != NEW; } public boolean cancel(boolean mayInterruptIfRunning) { if (!(state == NEW && UNSAFE.compareAndSwapInt(this, stateOffset, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED))) return false; try { // in case call to interrupt throws exception if (mayInterruptIfRunning) { try { Thread t = runner; if (t != null) t.interrupt(); } finally { // final state UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); } } } finally { finishCompletion(); } return true; } 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); } protected void done() { } protected void set(V v) { if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) { outcome = v; UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state finishCompletion(); } } ...
1》java.util.concurrent.Executors#callable(java.lang.Runnable, T) 采用适配器模式，将Runnable 适配为Callable
public static <T> Callable<T> callable(Runnable task, T result) { if (task == null) throw new NullPointerException(); return new RunnableAdapter<T>(task, result); }
java.util.concurrent.Executors.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; } }
可以看到是返回固定的结果， 也就是接受Runnable 的时候，返回的结果是固定的结果
2》然后同上面execute 方法一样，执行execute(ftask); 方法，这时候线程池跑的是FutureTask 对象
3》return ftask; 返回Future 对象，使得可以外部拿该对象可以获取执行结果。
4》线程跑任务会调用java.util.concurrent.FutureTask#run， 方法如下：
public void run() { 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; setException(ex); } if (ran) set(result); } } finally { // runner must be non-null until state is settled to // prevent concurrent calls to run() runner = null; // state must be re-read after nulling runner to prevent // leaked interrupts int s = state; if (s >= INTERRUPTING) handlePossibleCancellationInterrupt(s); } }
2. java.util.concurrent.AbstractExecutorService#submit(java.lang.Runnable, T) 理解
public <T> Future<T> submit(Runnable task, T result) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task, result); execute(ftask); return ftask; }
根据上面的理解，是返回固定的结果result， 只不过java.util.concurrent.AbstractExecutorService#submit(java.lang.Runnable) 的返回结果是null。
3. java.util.concurrent.AbstractExecutorService#submit(java.util.concurrent.Callable<T>) 理解
(1) 测试代码
Future<String> submit1 = threadPoolExecutor.submit(new Callable<String>() { @Override public String call() throws Exception { log.info("--" + Thread.currentThread().getName() + "--E"); try { Thread.sleep(2 * 1000); } catch (InterruptedException e) { } log.info("--" + Thread.currentThread().getName() + "--F"); return "123456"; } });
(2) 源码：java.util.concurrent.AbstractExecutorService#submit(java.util.concurrent.Callable<T>)
public <T> Future<T> submit(Callable<T> task) { if (task == null) throw new NullPointerException(); RunnableFuture<T> ftask = newTaskFor(task); execute(ftask); return ftask; }
1》java.util.concurrent.AbstractExecutorService#newTaskFor(java.util.concurrent.Callable<T>) 创建 FutureTask
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { return new FutureTask<T>(callable); }
接着调用：java.util.concurrent.FutureTask#FutureTask(java.util.concurrent.Callable<V>)
public FutureTask(Callable<V> callable) { if (callable == null) throw new NullPointerException(); this.callable = callable; this.state = NEW; // ensure visibility of callable }
2》然后调用java.util.concurrent.ThreadPoolExecutor#execute 跑任务，只不过Runnable 对象是java.util.concurrent.FutureTask， 内部的run 方法是调用成员属性Callable 的call 方法，方法执行完之后记录其返回结果， 所以FutureTask 可以拿到返回的结果。
补充：Future 两个重要的方法
(1) java.util.concurrent.FutureTask#isDone 判断是否执行完成
public boolean isDone() { return state != NEW; }
(2) 获取结果：
public V get() throws InterruptedException, ExecutionException { int s = state; if (s <= COMPLETING) s = awaitDone(false, 0L); return report(s); }
java.util.concurrent.FutureTask#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) // 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); } }
可以看到这里使用了 java.util.concurrent.locks.LockSupport#park(java.lang.Object) 进行阻塞线程，下面使用的是UNSAFE 类的相关API。
补充：关于LockSupport 的使用
参考：https://www.jianshu.com/p/8e6b1942ae39
测试代码：
package org.example; import java.util.concurrent.locks.LockSupport; public class PlainTest2 { public static void main(String[] args) { Thread t = new Thread(new Runnable() { @Override public void run() { System.out.println("111222"); LockSupport.park(); System.out.println("333444"); //LockSupport.park(obj); } }); t.start(); try { Thread.sleep(10000); } catch (InterruptedException e) { } System.out.println("555555"); LockSupport.unpark(t); System.out.println("555555"); } }
查看LockSupport 源码：
public static void park() { UNSAFE.park(false, 0L); } public static void unpark(Thread thread) { if (thread != null) UNSAFE.unpark(thread); }
strace linux环境下面监测如下：
1》park
clock_gettime(CLOCK_MONOTONIC, {tv_sec=273, tv_nsec=622979062}) = 0
futex(0xf6797ab8, FUTEX_WAIT_PRIVATE, 1, NULL) = 0
futex(0xf6797a9c, FUTEX_WAIT_PRIVATE, 2, NULL) = 0
futex(0xf6797a9c, FUTEX_WAKE_PRIVATE, 1) = 0
clock_gettime(CLOCK_MONOTONIC, {tv_sec=373, tv_nsec=621209093}) = 0
clock_gettime(CLOCK_MONOTONIC, {tv_sec=373, tv_nsec=621237803}) = 0
futex(0xf6797344, FUTEX_WAIT_PRIVATE, 1, {tv_sec=0, tv_nsec=971290}) = 0
futex(0xf6797328, FUTEX_WAIT_PRIVATE, 2, NULL) = 0
futex(0xf6797328, FUTEX_WAKE_PRIVATE, 1) = 0
2》unpark
write(1, "555555", 6) = 6 write(1, " ", 1) = 1 clock_gettime(CLOCK_MONOTONIC, {tv_sec=373, tv_nsec=620209800}) = 0 futex(0xf6797ab8, FUTEX_WAKE_OP_PRIVATE, 1, 1, 0xf6797ab4, FUTEX_OP_SET<<28|0<<12|FUTEX_OP_CMP_GT<<24|0x1) = 1 futex(0xf6797a9c, FUTEX_WAKE_PRIVATE, 1) = 1 write(1, "555555", 6) = 6
可以看到到内核调用的也是futex 指令，和synchronized 一样。
3. JDK1.8 的 CompletableFuture
java.util.concurrent.CompletableFuture 实现类实现了接口 Future、 CompletionStage。 Future 接口前面介绍过了，是一个异步的接口，提供了重要的isDone、get() 等方法。 CompletionStage 提供了一个链式操作的API，其重要方法如下：
1. Future 的局限性：
1. Future 的结果只能通过轮询判断isDone 或者 get 阻塞的方式获取结果
2. 多个 Future 不能串联在一起组成链式调用
3. 不能组合多个 Future 的结果
4. 没有异常处理
CompletableFuture 实现了 Future 和 CompletionStage接口，并且提供了许多关于创建，链式调用和组合多个 Future 的便利方法集，而且有广泛的异常处理支持。
CompletableFuture可以从全局的 ForkJoinPool.commonPool()获得一个线程中执行这些任务。但是你也可以创建一个线程池并传给该类的方法来让他们从线程池中获取一个线程执行它们的任务。CompletableFuture API 的所有方法都有两个变体-一个接受Executor作为参数。
可以用join阻塞获取结果，也可以用 get 阻塞获取结果。 区别是 get 会抛出受检查的异常 InterruptedException 和 ExecutionException。
2. 创建CompletableFuture
1. 简单的例子
package org.example; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; public class PlainTest2 { private static final Logger log = LoggerFactory.getLogger(PlainTest.class); public static void main(String[] args) { ExecutorService executor = Executors.newFixedThreadPool(5, new ThreadFactory() { private AtomicInteger atomicInteger = new AtomicInteger(0); private String threadNamePrefix = "myExecutor-"; @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r); thread.setName(threadNamePrefix + atomicInteger.incrementAndGet()); return thread; } }); // 1. 使用自定义的线程池 CompletableFuture<Void> rFuture = CompletableFuture .runAsync(() -> { log.info("hello 1 start"); try { Thread.sleep(5 * 1000); } catch (InterruptedException e) { } log.info("hello 1 end"); }, executor); // 阻塞等待，runAsync 的future 无返回值，输出null; 且join不会抛出异常 log.info(rFuture.join() + ""); // supplyAsync的使用 CompletableFuture<String> future = CompletableFuture .supplyAsync(() -> { log.info("hello 2 start"); try { Thread.sleep(5 * 1000); } catch (InterruptedException e) { } log.info("hello 2 end"); return "hello 2 result"; }); //阻塞等待 try { log.info(future.get()); } catch (InterruptedException e) { // ignore } catch (ExecutionException e) { // ignore } } }
结果：
2021-08-10 11:16:38 [myExecutor-1] [org.example.PlainTest]-[INFO] hello 1 start 2021-08-10 11:16:43 [myExecutor-1] [org.example.PlainTest]-[INFO] hello 1 end 2021-08-10 11:16:43 [main] [org.example.PlainTest]-[INFO] null 2021-08-10 11:16:43 [ForkJoinPool.commonPool-worker-1] [org.example.PlainTest]-[INFO] hello 2 start 2021-08-10 11:16:48 [ForkJoinPool.commonPool-worker-1] [org.example.PlainTest]-[INFO] hello 2 end 2021-08-10 11:16:48 [main] [org.example.PlainTest]-[INFO] hello 2 result
2. 源码查看：
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) { return asyncSupplyStage(asyncPool, supplier); } public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor) { return asyncSupplyStage(screenExecutor(executor), supplier); } public static CompletableFuture<Void> runAsync(Runnable runnable) { return asyncRunStage(asyncPool, runnable); } public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) { return asyncRunStage(screenExecutor(executor), runnable); }
(1)java.util.concurrent.CompletableFuture#asyncPool 是默认使用的线程池，默认使用Forkjoinpool.commonPool， 如果禁用就自己创建一个Execotor
private static final Executor asyncPool = useCommonPool ? ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
(2) java.util.concurrent.CompletableFuture#asyncSupplyStage 源码跟踪：
static <U> CompletableFuture<U> asyncSupplyStage(Executor e, Supplier<U> f) { if (f == null) throw new NullPointerException(); CompletableFuture<U> d = new CompletableFuture<U>(); e.execute(new AsyncSupply<U>(d, f)); return d; }
1》java.util.concurrent.CompletableFuture.AsyncSupply:
static final class AsyncSupply<T> extends ForkJoinTask<Void> implements Runnable, AsynchronousCompletionTask { CompletableFuture<T> dep; Supplier<T> fn; AsyncSupply(CompletableFuture<T> dep, Supplier<T> fn) { this.dep = dep; this.fn = fn; } public final Void getRawResult() { return null; } public final void setRawResult(Void v) {} public final boolean exec() { run(); return true; } public void run() { CompletableFuture<T> d; Supplier<T> f; if ((d = dep) != null && (f = fn) != null) { dep = null; fn = null; if (d.result == null) { try { d.completeValue(f.get()); } catch (Throwable ex) { d.completeThrowable(ex); } } d.postComplete(); } } }
继承Forkjointask， ForkJoinTask 又实现Future 接口。
2》可以看到该类的exec 方法调用run() 方法，run() 方法主要就是调用Supplier 的 get() 方法来进行操作，并且返回计算结果。
(3) java.util.concurrent.CompletableFuture#asyncRunStage：
static CompletableFuture<Void> asyncRunStage(Executor e, Runnable f) { if (f == null) throw new NullPointerException(); CompletableFuture<Void> d = new CompletableFuture<Void>(); e.execute(new AsyncRun(d, f)); return d; }
1》 java.util.concurrent.CompletableFuture.AsyncRun
static final class AsyncRun extends ForkJoinTask<Void> implements Runnable, AsynchronousCompletionTask { CompletableFuture<Void> dep; Runnable fn; AsyncRun(CompletableFuture<Void> dep, Runnable fn) { this.dep = dep; this.fn = fn; } public final Void getRawResult() { return null; } public final void setRawResult(Void v) {} public final boolean exec() { run(); return true; } public void run() { CompletableFuture<Void> d; Runnable f; if ((d = dep) != null && (f = fn) != null) { dep = null; fn = null; if (d.result == null) { try { f.run(); d.completeNull(); } catch (Throwable ex) { d.completeThrowable(ex); } } d.postComplete(); } } }
2》 可以看到设计同上面，只是AsyncRun.run 内部调用的是runabble.run， 然后记录结果为null。 结果记录在d 对象，也就是一开始创建的CompletableFuture 对象内部。
2. thenApply、thenAccept、thenRun 可以用于异步处理结果
package org.example; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; public class PlainTest2 { private static final Logger log = LoggerFactory.getLogger(PlainTest.class); public static void main(String[] args) { ExecutorService executor = Executors.newFixedThreadPool(5, new ThreadFactory() { private AtomicInteger atomicInteger = new AtomicInteger(0); private String threadNamePrefix = "myExecutor-"; @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r); thread.setName(threadNamePrefix + atomicInteger.incrementAndGet()); return thread; } }); // 1. 使用自定义的线程池 CompletableFuture<Void> rFuture = CompletableFuture .runAsync(() -> { log.info("hello 1 start"); try { Thread.sleep(5 * 1000); } catch (InterruptedException e) { } log.info("hello 1 end"); }, executor).thenRunAsync(() -> { // 继续另一个任务 log.info("hello 11 start"); try { Thread.sleep(5 * 1000); } catch (InterruptedException e) { } log.info("hello 11 end"); }, executor); // 阻塞等待，runAsync 的future 无返回值，输出null; 且join不会抛出异常 log.info(rFuture.join() + ""); // thenApply、thenAccept、thenRun 可以用于异步处理结果 CompletableFuture<String> future = CompletableFuture .supplyAsync(() -> { log.info("hello 2 start"); try { Thread.sleep(5 * 1000); } catch (InterruptedException e) { } log.info("hello 2 end"); return "hello 2 result"; }).thenApply(result -> { // 对结果进行转换，可以连续转换 log.info("result: {}", result); return "hello 2 result thenApply"; }); //阻塞等待 try { log.info(future.get()); } catch (InterruptedException e) { // ignore } catch (ExecutionException e) { // ignore } } }
结果：
2021-08-10 12:48:34 [myExecutor-1] [org.example.PlainTest]-[INFO] hello 1 start 2021-08-10 12:48:39 [myExecutor-1] [org.example.PlainTest]-[INFO] hello 1 end 2021-08-10 12:48:39 [myExecutor-2] [org.example.PlainTest]-[INFO] hello 11 start 2021-08-10 12:48:44 [myExecutor-2] [org.example.PlainTest]-[INFO] hello 11 end 2021-08-10 12:48:44 [main] [org.example.PlainTest]-[INFO] null 2021-08-10 12:48:44 [ForkJoinPool.commonPool-worker-1] [org.example.PlainTest]-[INFO] hello 2 start 2021-08-10 12:48:49 [ForkJoinPool.commonPool-worker-1] [org.example.PlainTest]-[INFO] hello 2 end 2021-08-10 12:48:49 [ForkJoinPool.commonPool-worker-1] [org.example.PlainTest]-[INFO] result: hello 2 result 2021-08-10 12:48:49 [main] [org.example.PlainTest]-[INFO] hello 2 result thenApply
3. 合并两个CompletableFuture
测试代码：
package org.example; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiFunction; import java.util.function.Function; public class PlainTest2 { private static final Logger log = LoggerFactory.getLogger(PlainTest.class); public static void main(String[] args) { ExecutorService executor = Executors.newFixedThreadPool(5, new ThreadFactory() { private AtomicInteger atomicInteger = new AtomicInteger(0); private String threadNamePrefix = "myExecutor-"; @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r); thread.setName(threadNamePrefix + atomicInteger.incrementAndGet()); return thread; } }); CompletableFuture<String> future = CompletableFuture .supplyAsync(() -> { log.info("hello 1 start"); try { Thread.sleep(5 * 1000); } catch (InterruptedException e) { } log.info("hello 1 end"); return "hello 1 result"; }, executor); CompletableFuture<String> future2 = CompletableFuture.completedFuture(" OK"); // 常量任务 // 第一种合并结果 CompletableFuture<String> stringCompletableFuture = future.thenComposeAsync(new Function<String, CompletionStage<String>>() { @Override public CompletionStage<String> apply(String s) { log.info("s: {}", s); return CompletableFuture.completedFuture(s + future2.join()); } }); // 第二种合并： 当两个Future都完成的时候，传给``thenCombine()的回调函数将被调用。 CompletableFuture<Object> objectCompletableFuture = future.thenCombine(future2, new BiFunction<String, String, Object>() { @Override public Object apply(String s, String s2) { return s + s2; } }); //阻塞等待 try { log.info(stringCompletableFuture.get()); log.info(objectCompletableFuture.get() + ""); } catch (InterruptedException e) { // ignore } catch (ExecutionException e) { // ignore } } }
结果：
2021-08-10 13:56:50 [myExecutor-1] [org.example.PlainTest]-[INFO] hello 1 start 2021-08-10 13:56:55 [myExecutor-1] [org.example.PlainTest]-[INFO] hello 1 end 2021-08-10 13:56:55 [ForkJoinPool.commonPool-worker-1] [org.example.PlainTest]-[INFO] s: hello 1 result 2021-08-10 13:56:55 [main] [org.example.PlainTest]-[INFO] hello 1 result OK 2021-08-10 13:56:55 [main] [org.example.PlainTest]-[INFO] hello 1 result OK
4. 组合多个CompletableFuture
allof、anyOf 用于组合多个CompletableFuture，区别是allOf(多个任务都需要执行完, 返回的结果类型是CompletableFuture<Void>， 也就是不接受返回结果)、anyOf(有一个任务执行完即可, 返回值类型是CompletableFuture<Object>)
测试代码：
package org.example; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; public class PlainTest2 { private static final Logger log = LoggerFactory.getLogger(PlainTest.class); public static void main(String[] args) { ExecutorService executor = Executors.newFixedThreadPool(5, new ThreadFactory() { private AtomicInteger atomicInteger = new AtomicInteger(0); private String threadNamePrefix = "myExecutor-"; @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r); thread.setName(threadNamePrefix + atomicInteger.incrementAndGet()); return thread; } }); CompletableFuture<String> future = CompletableFuture .supplyAsync(() -> { log.info("hello 1 start"); try { Thread.sleep(5 * 1000); } catch (InterruptedException e) { } log.info("hello 1 end"); return "hello 1 result"; }, executor); CompletableFuture<String> future2 = CompletableFuture .supplyAsync(() -> { log.info("hello 2 start"); try { Thread.sleep(10 * 1000); } catch (InterruptedException e) { } log.info("hello 2 end"); return "hello 2 result"; }, executor); // 多个任务的组合： allOf(多个任务都需要执行完, 返回的结果类型是CompletableFuture<Void>， 也就是不接受返回结果)、anyOf(有一个任务执行完即可, 返回值类型是CompletableFuture<Object>) CompletableFuture<Object> voidCompletableFuture = CompletableFuture.anyOf(future, future2); try { // 阻塞等待 log.info(voidCompletableFuture.get() + ""); } catch (InterruptedException e) { // ignore } catch (ExecutionException e) { // ignore } } }
结果：
2021-08-10 14:36:58 [myExecutor-2] [org.example.PlainTest]-[INFO] hello 2 start 2021-08-10 14:36:58 [myExecutor-1] [org.example.PlainTest]-[INFO] hello 1 start 2021-08-10 14:37:03 [myExecutor-1] [org.example.PlainTest]-[INFO] hello 1 end 2021-08-10 14:37:03 [main] [org.example.PlainTest]-[INFO] hello 1 result 2021-08-10 14:37:08 [myExecutor-2] [org.example.PlainTest]-[INFO] hello 2 end
5. 异常处理
方法一：exceptionally-可以在这里记录这个异常并返回一个默认值
package org.example; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; public class PlainTest2 { private static final Logger log = LoggerFactory.getLogger(PlainTest.class); public static void main(String[] args) { ExecutorService executor = Executors.newFixedThreadPool(5, new ThreadFactory() { private AtomicInteger atomicInteger = new AtomicInteger(0); private String threadNamePrefix = "myExecutor-"; @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r); thread.setName(threadNamePrefix + atomicInteger.incrementAndGet()); return thread; } }); CompletableFuture future = CompletableFuture.supplyAsync(() -> { log.info("======1======"); return "111222"; }, executor).thenApply(result -> { log.info("======2======"); int i = 1 / 0; return "222333"; }).thenApply(result -> { log.info("======3======"); return "333444"; }).exceptionally(exception -> { exception.printStackTrace(); return exception.getMessage(); }); try { // 阻塞等待 log.info(future.get() + ""); } catch (InterruptedException e) { // ignore } catch (ExecutionException e) { // ignore } } }
结果：
2021-08-10 15:04:31 [myExecutor-1] [org.example.PlainTest]-[INFO] ======1====== 2021-08-10 15:04:31 [myExecutor-1] [org.example.PlainTest]-[INFO] ======2====== 2021-08-10 15:04:31 [main] [org.example.PlainTest]-[INFO] java.lang.ArithmeticException: / by zero java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273) at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280) at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:618) at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591) at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488) at java.util.concurrent.CompletableFuture$AsyncSupply.run$$$capture(CompletableFuture.java:1609) at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Caused by: java.lang.ArithmeticException: / by zero at org.example.PlainTest2.lambda$main$1(PlainTest2.java:31) at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616) ... 7 more
方法二： handle - 相比exceptionally而言，即可处理上一环节的异常也可以处理其正常返回值， 也就是异常或者不异常都会走handle
测试代码：
package org.example; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; public class PlainTest2 { private static final Logger log = LoggerFactory.getLogger(PlainTest.class); public static void main(String[] args) { ExecutorService executor = Executors.newFixedThreadPool(5, new ThreadFactory() { private AtomicInteger atomicInteger = new AtomicInteger(0); private String threadNamePrefix = "myExecutor-"; @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r); thread.setName(threadNamePrefix + atomicInteger.incrementAndGet()); return thread; } }); CompletableFuture future = CompletableFuture.supplyAsync(() -> { log.info("======1======"); return "111222"; }, executor).thenApply(result -> { log.info("======2======"); return "222333"; }).thenApply(result -> { log.info("======3======"); return "333444"; }).handle((data, exception) -> { log.info("data: {}", data); if (exception == null) { return "444555"; } exception.printStackTrace(); return exception.getMessage(); }); try { // 阻塞等待 log.info(future.get() + ""); } catch (InterruptedException e) { // ignore } catch (ExecutionException e) { // ignore } } }
结果：
2021-08-10 15:25:17 [myExecutor-1] [org.example.PlainTest]-[INFO] ======1====== 2021-08-10 15:25:17 [myExecutor-1] [org.example.PlainTest]-[INFO] ======2====== 2021-08-10 15:25:17 [myExecutor-1] [org.example.PlainTest]-[INFO] ======3====== 2021-08-10 15:25:17 [myExecutor-1] [org.example.PlainTest]-[INFO] data: 333444 2021-08-10 15:25:21 [main] [org.example.PlainTest]-[INFO] 444555
方法三：whenComplete - 它不参与返回结果的处理，把它当成监听器即可
package org.example; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; public class PlainTest2 { private static final Logger log = LoggerFactory.getLogger(PlainTest.class); public static void main(String[] args) { ExecutorService executor = Executors.newFixedThreadPool(5, new ThreadFactory() { private AtomicInteger atomicInteger = new AtomicInteger(0); private String threadNamePrefix = "myExecutor-"; @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r); thread.setName(threadNamePrefix + atomicInteger.incrementAndGet()); return thread; } }); CompletableFuture future = CompletableFuture.supplyAsync(() -> { log.info("======1======"); return "111222"; }, executor).thenApply(result -> { log.info("======2======"); int i = 1 / 0; return "222333"; }).thenApply(result -> { log.info("======3======"); return "333444"; }).whenComplete((data, exception) -> { // 它不参与返回结果的处理，把它当成监听器即可 log.info("data: {}", data); if (exception != null) { exception.printStackTrace(); } }); try { // 阻塞等待 log.info(future.get() + ""); } catch (InterruptedException e) { // ignore } catch (ExecutionException e) { // ignore } } }
结果：
2021-08-10 15:49:32 [myExecutor-1] [org.example.PlainTest]-[INFO] ======1====== 2021-08-10 15:49:32 [myExecutor-1] [org.example.PlainTest]-[INFO] ======2====== 2021-08-10 15:49:32 [myExecutor-1] [org.example.PlainTest]-[INFO] data: null java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273) at java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:280) at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:618) at java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:591) at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488) at java.util.concurrent.CompletableFuture$AsyncSupply.run$$$capture(CompletableFuture.java:1609) at java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) at java.lang.Thread.run(Thread.java:748) Caused by: java.lang.ArithmeticException: / by zero at org.example.PlainTest2.lambda$main$1(PlainTest2.java:31) at java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616) ... 7 more
补充： 创建一个空的CompletableFuture， 其isDone() 是false，get() 方法会阻塞
// 默认创建一个空的 CompletableFuture ， isDone 是false CompletableFuture completableFuture = new CompletableFuture(); System.out.println(completableFuture.isDone()); // 完成该任务，并将结果设为 123 completableFuture.complete(123); System.out.println(completableFuture.isDone()); System.out.println(completableFuture.get());
结果：
false
true
123
java.util.concurrent.CompletableFuture#isDone 方法判断是否完成的依据是根据是否有结果进行判断：
public boolean isDone() { return result != null; }