java线程池ThreadPoolExecutor通过submit提交runnable的task时,为何不会抛出oom

缺乏、安全感 2023-06-25 06:07 7阅读 0赞

## 背景介绍 ##

最近在听朋友说threadlocal会存在内存泄露，是因为它的entry是弱引用导致的，然后我也自己百度了一下，看了下它存存泄露的原因。

但通过网上的文章描述中，很难看到对threadlocal会导致内存泄露有相关的示例代码，而且还有很多文章都是你抄我、我抄你、互相借鉴的非常多。于是，我便想着自己来写一个示例代码。

threadlocal会造成内存泄露最简单的方法就是写一个线程池，线程不会回收，那么threadlocal对应map中的entry中的value也就不会回收，因为thread对于threadlocal的引用链会一直存在。

### runnable通过线程池submit提交不oom重现代码 ###

import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class ThreadLocalOomTest {
    
        public static void main(String[] args) {
            ExecutorService executorService = Executors.newFixedThreadPool(1);
            for (int i = 0; i < 500; i++) {
                int finalI = i;
                executorService.submit(new Runnable() {
                    @Override
                    public void run() {
                        System.out.println("now,pre run" + Thread.currentThread().toString() + ":" + finalI);
                        try {
                            Thread.sleep(10_000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        ThreadLocal<Byte[]> threadLocal = new ThreadLocal<>();
                        Byte[] bytes = new Byte[1024 * 1024 * 9];
                        threadLocal.set(bytes);
                        System.out.println("----------------->" + Thread.currentThread().toString() + ":" + finalI);
                        if (finalI == 499) {
                            executorService.shutdown();
                        }
                        System.out.println(finalI+" run");
                    }
                });
            }
        }
    }

为了测试的方便，我特意将最大堆的大小设置为了128M

> \-Xmx128m

其输出结果感觉很奇怪，前几条输出如下：

观察上面的输出片段，会发现只有0、1、6运行成功了，但并没有抛出oom异常信息。再观察下堆情况

![gc堆信息][gc]

但如果将上面提交到线程池的测试代码换成execute，其输出情况如下：

now,pre runThread[pool-1-thread-1,5,main]:0
    ----------------->Thread[pool-1-thread-1,5,main]:0
    0 run
    now,pre runThread[pool-1-thread-1,5,main]:1
    ----------------->Thread[pool-1-thread-1,5,main]:1
    1 run
    now,pre runThread[pool-1-thread-1,5,main]:2
    Exception in thread "pool-1-thread-1" now,pre runThread[pool-1-thread-2,5,main]:3
    java.lang.OutOfMemoryError: Java heap space
    	at com.example.javastu.threadlocaltest.ThreadLocalOomTest$1.run(ThreadLocalOomTest.java:22)
    	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)
    ----------------->Thread[pool-1-thread-2,5,main]:3
    3 run
    now,pre runThread[pool-1-thread-2,5,main]:4
    ----------------->Thread[pool-1-thread-2,5,main]:4
    4 run
    now,pre runThread[pool-1-thread-2,5,main]:5
    ----------------->Thread[pool-1-thread-2,5,main]:5
    5 run
    now,pre runThread[pool-1-thread-2,5,main]:6
    ----------------->Thread[pool-1-thread-2,5,main]:6
    6 run
    now,pre runThread[pool-1-thread-2,5,main]:7
    ----------------->Thread[pool-1-thread-2,5,main]:7
    7 run
    now,pre runThread[pool-1-thread-2,5,main]:8
    ----------------->Thread[pool-1-thread-2,5,main]:8
    8 run
    now,pre runThread[pool-1-thread-2,5,main]:9
    ----------------->Thread[pool-1-thread-2,5,main]:9
    9 run
    now,pre runThread[pool-1-thread-2,5,main]:10
    Exception in thread "pool-1-thread-2" java.lang.OutOfMemoryError: Java heap space
    	at com.example.javastu.threadlocaltest.ThreadLocalOomTest$1.run(ThreadLocalOomTest.java:22)
    	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)
    now,pre runThread[pool-1-thread-3,5,main]:11
    ----------------->Thread[pool-1-thread-3,5,main]:11
    11 run
    now,pre runThread[pool-1-thread-3,5,main]:12
    ----------------->Thread[pool-1-thread-3,5,main]:12
    12 run

当用execute方法提交后，又会报出oom异常了。  
先抛开输出的具体细节，只看submit和execute方法就感觉有一些疑惑的地方：都是在线程池中运行，为啥输出结果就不一样了呢？

为了了解其原理，我决定还是看一下源码,了解下execute为啥可能抛出oom，而submit不会抛出oom的原因

## ThreadPoolExecutor execute方法 ##

/**
         * Executes the given task sometime in the future.  The task
         * may execute in a new thread or in an existing pooled thread.
         *
         * If the task cannot be submitted for execution, either because this
         * executor has been shutdown or because its capacity has been reached,
         * the task is handled by the current {@code RejectedExecutionHandler}.
         *
         * @param command the task to execute
         * @throws RejectedExecutionException at discretion of
         *         {@code RejectedExecutionHandler}, if the task
         *         cannot be accepted for execution
         * @throws NullPointerException if {@code command} is null
         */
        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);
        }

正常情况下，会再进入addWorker方法,其代码如下：

/*
         * Methods for creating, running and cleaning up after workers
         */
    
        /**
         * Checks if a new worker can be added with respect to current
         * pool state and the given bound (either core or maximum). If so,
         * the worker count is adjusted accordingly, and, if possible, a
         * new worker is created and started, running firstTask as its
         * first task. This method returns false if the pool is stopped or
         * eligible to shut down. It also returns false if the thread
         * factory fails to create a thread when asked.  If the thread
         * creation fails, either due to the thread factory returning
         * null, or due to an exception (typically OutOfMemoryError in
         * Thread.start()), we roll back cleanly.
         *
         * @param firstTask the task the new thread should run first (or
         * null if none). Workers are created with an initial first task
         * (in method execute()) to bypass queuing when there are fewer
         * than corePoolSize threads (in which case we always start one),
         * or when the queue is full (in which case we must bypass queue).
         * Initially idle threads are usually created via
         * prestartCoreThread or to replace other dying workers.
         *
         * @param core if true use corePoolSize as bound, else
         * maximumPoolSize. (A boolean indicator is used here rather than a
         * value to ensure reads of fresh values after checking other pool
         * state).
         * @return true if successful
         */
        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对象，将先会将传入的task传入到新建的Worker中,然后再从Worker中拿到这个Worker的thread变量，再调用了这个Worker的thread的start方法，那么start()方法运行的是哪里的代码呢？

再来看一下Worker的源码：

/**
         * Class Worker mainly maintains interrupt control state for
         * threads running tasks, along with other minor bookkeeping.
         * This class opportunistically extends AbstractQueuedSynchronizer
         * to simplify acquiring and releasing a lock surrounding each
         * task execution.  This protects against interrupts that are
         * intended to wake up a worker thread waiting for a task from
         * instead interrupting a task being run.  We implement a simple
         * non-reentrant mutual exclusion lock rather than use
         * ReentrantLock because we do not want worker tasks to be able to
         * reacquire the lock when they invoke pool control methods like
         * setCorePoolSize.  Additionally, to suppress interrupts until
         * the thread actually starts running tasks, we initialize lock
         * state to a negative value, and clear it upon start (in
         * runWorker).
         */
        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;
    
            /**
             * Creates with given first task and thread from ThreadFactory.
             * @param firstTask the first task (null if none)
             */
            Worker(Runnable firstTask) {
                setState(-1); // inhibit interrupts until runWorker
                this.firstTask = firstTask;
                this.thread = getThreadFactory().newThread(this);
            }
    
            /** Delegates main run loop to outer runWorker  */
            public void run() {
                runWorker(this);
            }
    
            // Lock methods
            //
            // The value 0 represents the unlocked state.
            // The value 1 represents the locked state.
    
            protected boolean isHeldExclusively() {
                return getState() != 0;
            }
    
            protected boolean tryAcquire(int unused) {
                if (compareAndSetState(0, 1)) {
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
                return false;
            }
    
            protected boolean tryRelease(int unused) {
                setExclusiveOwnerThread(null);
                setState(0);
                return true;
            }
    
            public void lock()        { acquire(1); }
            public boolean tryLock()  { return tryAcquire(1); }
            public void unlock()      { release(1); }
            public boolean isLocked() { return isHeldExclusively(); }
    
            void interruptIfStarted() {
                Thread t;
                if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
                    try {
                        t.interrupt();
                    } catch (SecurityException ignore) {
                    }
                }
            }
        }

观察上面的这部分：

/**
             * Creates with given first task and thread from ThreadFactory.
             * @param firstTask the first task (null if none)
             */
            Worker(Runnable firstTask) {
                setState(-1); // inhibit interrupts until runWorker
                this.firstTask = firstTask;
                this.thread = getThreadFactory().newThread(this);
            }

在新建Worker的时候，会将Worker中的thread也一并创建好，并将Worker自己塞到了自己 的Thread中，因为Worker实现了Runnable接口

所以上面t.start()方法实际上运行的也就是Worker中的run()方法，Worker的run方法调用的是runWorker，其代码如下：

/**
         * Main worker run loop.  Repeatedly gets tasks from queue and
         * executes them, while coping with a number of issues:
         *
         * 1. We may start out with an initial task, in which case we
         * don't need to get the first one. Otherwise, as long as pool is
         * running, we get tasks from getTask. If it returns null then the
         * worker exits due to changed pool state or configuration
         * parameters.  Other exits result from exception throws in
         * external code, in which case completedAbruptly holds, which
         * usually leads processWorkerExit to replace this thread.
         *
         * 2. Before running any task, the lock is acquired to prevent
         * other pool interrupts while the task is executing, and then we
         * ensure that unless pool is stopping, this thread does not have
         * its interrupt set.
         *
         * 3. Each task run is preceded by a call to beforeExecute, which
         * might throw an exception, in which case we cause thread to die
         * (breaking loop with completedAbruptly true) without processing
         * the task.
         *
         * 4. Assuming beforeExecute completes normally, we run the task,
         * gathering any of its thrown exceptions to send to afterExecute.
         * We separately handle RuntimeException, Error (both of which the
         * specs guarantee that we trap) and arbitrary Throwables.
         * Because we cannot rethrow Throwables within Runnable.run, we
         * wrap them within Errors on the way out (to the thread's
         * UncaughtExceptionHandler).  Any thrown exception also
         * conservatively causes thread to die.
         *
         * 5. After task.run completes, we call afterExecute, which may
         * also throw an exception, which will also cause thread to
         * die. According to JLS Sec 14.20, this exception is the one that
         * will be in effect even if task.run throws.
         *
         * The net effect of the exception mechanics is that afterExecute
         * and the thread's UncaughtExceptionHandler have as accurate
         * information as we can provide about any problems encountered by
         * user code.
         *
         * @param w the worker
         */
        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);
            }
        }

看了一下runWorker代码后，就进入线程池运行task的关键代码了

> task.run()

task.run方法便就是task运行的关键代码，通过观察其代码可以发现，task.run()方法是有层层异常捕获的，那么如果有oom异常时，就会被其中的Error异常捕获起来，并抛出来,因为oom异常是继承与Error的，其继承图如下：  
![oom类继承关系][oom]

所以，execute方法之所以会抛出oom异常的原因，我们通过上面的分析就已经知道了

## ThreadPoolExecutor submit方法 ##

接着，再来了解下线程池的submit方法，看下submit是如何对task运行过程中的异常是如何处理的

还是先从submit方法入口处开始看

/**
         * @throws RejectedExecutionException {@inheritDoc}
         * @throws NullPointerException       {@inheritDoc}
         */
        public Future<?> submit(Runnable task) {
            if (task == null) throw new NullPointerException();
            RunnableFuture<Void> ftask = newTaskFor(task, null);
            execute(ftask);
            return ftask;
        }

进入newTaskFor方法

/**
         * Returns a {@code RunnableFuture} for the given runnable and default
         * value.
         *
         * @param runnable the runnable task being wrapped
         * @param value the default value for the returned future
         * @param <T> the type of the given value
         * @return a {@code RunnableFuture} which, when run, will run the
         * underlying runnable and which, as a {@code Future}, will yield
         * the given value as its result and provide for cancellation of
         * the underlying task
         * @since 1.6
         */
        protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
            return new FutureTask<T>(runnable, value);
        }

newTaskFor会新建一个FutureTask,但大家都知道FutureTask是用来包装callable对象的，那么这里传人的是一个runnable的对象，它会怎么处理呢？

/**
         * Creates a {@code FutureTask} that will, upon running, execute the
         * given {@code Runnable}, and arrange that {@code get} will return the
         * given result on successful completion.
         *
         * @param runnable the runnable task
         * @param result the result to return on successful completion. If
         * you don't need a particular result, consider using
         * constructions of the form:
         * {@code Future<?> f = new FutureTask<Void>(runnable, null)}
         * @throws NullPointerException if the runnable is null
         */
        public FutureTask(Runnable runnable, V result) {
            this.callable = Executors.callable(runnable, result);
            this.state = NEW;       // ensure visibility of callable
        }

它会调用Executors.callable方法

/**
         * Returns a {@link Callable} object that, when
         * called, runs the given task and returns the given result.  This
         * can be useful when applying methods requiring a
         * {@code Callable} to an otherwise resultless action.
         * @param task the task to run
         * @param result the result to return
         * @param <T> the type of the result
         * @return a callable object
         * @throws NullPointerException if task null
         */
        public static <T> Callable<T> callable(Runnable task, T result) {
            if (task == null)
                throw new NullPointerException();
            return new RunnableAdapter<T>(task, result);
        }
    
    
        /**
         * A callable that runs given task and returns given result
         */
        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;
            }
        }

Executors.callable的方法最终会调用一个RunnableAdapter，通过这个适配器它会返回一个Callable的对象

那么Callable、RunnableAdapter、FutureTask这三个类有什么关系呢？我们来看一下继承关系图  
FutureTask的继承关系图如下：

![FutureTask继承关系][FutureTask]

RunnableAdapter的继承关系图如下：  
![RunnableAdapter继承关系][RunnableAdapter]

通过查看继承关系可知道，FutureTask与RunnableAdapter都实现了Callable接口，根据多态的原则，那么RunnableAdapter用FutureTask去接收也是可以的

源码看到这里，大家应该都知道了线程池中的submit方法如何将一个Runnable的task转为FutureTask了，那么Java线程池又是如何执行这个FutureTask的呢？

继续看刚开始的submit方法：

public Future<?> submit(Runnable task) {
            if (task == null) throw new NullPointerException();
            RunnableFuture<Void> ftask = newTaskFor(task, null);
            execute(ftask);
            return ftask;
        }

通过代码可知，在将runnable的task转为了RunnableFuture后，由execute方法进行执行了

仔细看一下上面的代码

> java.util.concurrent.ThreadPoolExecutor\#execute

仔细看一下之后就会发现，这就和上面分析的execute提交runnable的代码一样的，就是同一个方法

通过上面的分析我们已近知道了它的处理流程大致是这样的：  
它会先执行一个addWorker方法，然后会新建一个Worker，然后会执行Worker里的runWorker方法,通过runWorker方法会去执行原来传进去的task中的run方法

我们知道在runnable是通过submit提交时，那么这个runnable对象就是FutureTask,那么runWorker最终执行的也就是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);
            }
        }

通过上面的代码可知，如果task在运行过程中有异常抛出，会先在FutureTask里被先捕获起来,然后再通过

> setException(ex);

方法进行设置,setException源码如下：

/**
         * Causes this future to report an {@link ExecutionException}
         * with the given throwable as its cause, unless this future has
         * already been set or has been cancelled.
         *
         * <p>This method is invoked internally by the {@link #run} method
         * upon failure of the computation.
         *
         * @param t the cause of failure
         */
        protected void setException(Throwable t) {
            if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
                outcome = t;
                UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
                finishCompletion();
            }
        }

最终，原来的异常信息就变成outcome的值了,所以当Worker想再像以前去捕获异常的时候，就根本捕获不到异常的，因为FutureTask早已抢先一步

那么如何获取outcome的值呢？  
在源码里搜索outcome的关联地方即可，会找到report方法

/**
         * Returns result or throws exception for completed task.
         *
         * @param s completed state value
         */
        @SuppressWarnings("unchecked")
        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);
        }

再通过report方法寻找调用它的方法，最终会找到get方法

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

所以看到这里，我们便知道，对于使用submit提交的任务如果想要获取它运行过程中的异常，应该用get()方法去接收

## 总结 ##

来简短总结一下

ThreadPoolExecutor如果传入的task是runnable的话，如果让它抛出异常，就用execute方法提交

如果是用submit方法提交的，那么就用futureTask的get方法去接收

那么对于本文的原始问题：

问题：  
**java线程池ThreadPoolExecutor通过submit提交runnable时，为何不会抛出oom?**

回答：因为如果使用submit提交的话，如果task中的异常，会被FutureTask先进行捕获

[gc]: https://img-blog.csdnimg.cn/2020012014313794.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9wdWhhaXlhbmcuYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[oom]: https://img-blog.csdnimg.cn/20200120143215830.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9wdWhhaXlhbmcuYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[FutureTask]: https://img-blog.csdnimg.cn/20200120143250685.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9wdWhhaXlhbmcuYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70
[RunnableAdapter]: https://img-blog.csdnimg.cn/20200120143315991.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9wdWhhaXlhbmcuYmxvZy5jc2RuLm5ldA==,size_16,color_FFFFFF,t_70