Java Callable Future接口执行机制解密

阳光穿透心脏的1/2处 2022-06-07 02:38 229阅读 0赞

在Java中,我们执行异步任务的代码可以这么写。

ExecutorService executorService = Executors.newSingleThreadExecutor();
    
            Future<String> stringFuture = executorService.submit(() -> {
                System.out.println("你好,世界");
                return "helloworld";
            });
    
            System.out.println(stringFuture.getClass().getName());
            System.out.println(stringFuture.get());

在submit方法中传入一个callable接口对象。里面承载的就是异步执行的Java代码。 我们可以看到异步执行的任务很简单,打印一行话 你好,世界。然后将helloworld字符串返回给Future<T>接口对象接受执行的结果。当我们执行   stringFuture.get()方法的时候,如果异步的任务还没有执行完。那么future.get()方法所在的线程是会阻塞的。

像我们平常自己起一个线程去执行,执行的结果一般是那不会来的。但是Callable接口和Future接口这两个接口却可以做到将线程执行完的计算结果给拿到。

所以带着这样一个问题我们去查看下jdk的源码是怎样实现的。

首先第一行代码

ExecutorService executorService = Executors.newSingleThreadExecutor();

我们可以知道,创建了一个线程池。具体的代码是这样的。

public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }Executors类新建了一个ExecutorService接口的实现类。 
     
     
     该类的继承结构是这样的。 
     其中新建的ExecutorService接口的实现类主要使用的是DelegatedExecutorService这个类。该类中封装又封装了一个ExecutorService接口的实现类。 static class DelegatedExecutorService extends AbstractExecutorService {
        private final ExecutorService e;
        DelegatedExecutorService(ExecutorService executor) { e = executor; }
        public void execute(Runnable command) { e.execute(command); }
        public void shutdown() { e.shutdown(); }
        public List<Runnable> shutdownNow() { return e.shutdownNow(); }
        public boolean isShutdown() { return e.isShutdown(); }
        public boolean isTerminated() { return e.isTerminated(); }
        public boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException {
            return e.awaitTermination(timeout, unit);
        }
        public Future<?> submit(Runnable task) {
            return e.submit(task);
        }
        public <T> Future<T> submit(Callable<T> task) {
            return e.submit(task);
        }
        public <T> Future<T> submit(Runnable task, T result) {
            return e.submit(task, result);
        }
        public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
            throws InterruptedException {
            return e.invokeAll(tasks);
        }
        public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                             long timeout, TimeUnit unit)
            throws InterruptedException {
            return e.invokeAll(tasks, timeout, unit);
        }
        public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
            throws InterruptedException, ExecutionException {
            return e.invokeAny(tasks);
        }
        public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                               long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException {
            return e.invokeAny(tasks, timeout, unit);
        }
    }
    
      
     
    
    我们可以看到,该类的所有的方法实现全都是调用可内部封装的ExecutorService接口的实现类。
     
     有读者可能会问了,这个被包装的接口的类是哪个类。这里应该是ThreadPoolExecutor这个类。 我们可以看下集成关系图 
     ThraedPoolExecutor类也是继承了AbstractExecutorService这个类。这就好分析了。 接下来我们分析下面的代码。 以后笔者直接将分析的内容以注解的形式写在代码旁边 Future<String> stringFuture = executorService.submit(() -> {
        System.out.println("你好,世界");
        return "helloworld";
    });
     
    
     public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException(); //实际上这里是new了一个FutureTask类的对象。 //FutureTask类实现了Runnable,Future<T>和RunnableFuture<V>三个接口,说明之后返回的FutureTask类对象可以 //当做Future<T>接口的实现类来用
        RunnableFuture<T> ftask = newTaskFor(task); 	//Callable<T>接口里的异步代码是在这行代码被调用的。 	//execute方法接收了刚才那个FutureTask类的对象做为参数去执行。我们到这个execute()方法里去看看
        execute(ftask);
        return ftask;
    }
    
    
    
     
     我们可以看到该方法实现在ThreadPoolExecutor类里面。当然execute(Runnable r)这个方法是声明在Executor接口里面的。 好,我们到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);
    }
    
    
     
     //这个方法的形参就是刚才new出来的FutureTask类的对象。因为其也实现了Runnable接口,所以可以这样传递 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) {
            
     		//这行代码启动了Callable接口里的线程代码。
     		//既然线程被启动了,里面我们就要去看看Runnable接口定义的run方法是如何实现的,我们进入到FutureTask类的Run方法实现类中去
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }
    
    
     
     
     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 {
             
     		//这行代码调用了Callable接口实现类的call方法。得到的返回值会返回给一个result变量
     		//下面的代码将这个变量设置到FutureTask对象的成员变量outcome上。
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex);
                }
                if (ran)
     		//就是这行代码将Callable接口的call方法的执行结果设置到FutureTask类对象的outcome成员变量上
     		//我们可以点进入看一下是如何实现的
                    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);
        }
    }
    
    
     
     protected void set(V v) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
              
     	//将Callable接口的call方法的计算结果放置到outcome变量上。
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state finishCompletion();
        }
    }
    
    
     
     
     到这里我们可以知道,执行完的结果是保存在FutureTask类对象的outcome成员变量上的。 我们再去看下Future<T>接口定义的get方法 /** * @throws CancellationException { @inheritDoc} */ public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
     	//发现这里有个report方法,点进去看一下
        return report(s);
    }
     @SuppressWarnings("unchecked")
    private V report(int s) throws ExecutionException {
               //我们可以发现果然拿的是outcome成员变量的值
        Object x = outcome;
        if (s == NORMAL)
            return (V)x;
        if (s >= CANCELLED)
            throw new CancellationException();
        throw new ExecutionException((Throwable)x);
    }
    
    到此文章的结束了