java Future 阻塞

Posted 2023-03-24

比如有一条线程在每次调用一个方法：通过经纬度获取地理位置信息
如果我想大量的使用多线程并发
但在方法中调用了Future 的get方法，这样是不是阻塞了整条线程
和单一的线程调用没有啥区别？

有区别。比如你要查3次，每次分别耗时1s，2s，3s，用单线程，这个线程就会阻塞3次，总共耗时等于这3次查询的总耗时，是6s；而如果用了3个线程来查，每个线程都用Future 的get方法来返回数据，这样就是3个阻塞发生在同一时间,前两个完成了会继续等待第三个查询完成，最终耗时是时间最长的那个查询，是3s。追问

Future 的get不也产生阻塞吗？

Future 的get不也产生阻塞吗？

追答

是产生阻塞，但第一个线程在阻塞的时候，cpu可以去跑第二个、第三个的线程。打个比方，单线程是你一个人去三家店买东西，在每家店都要等老板拿货；而多线程是三个人分别去三家店买东西，虽然每个人在每家店都要等老板拿货，但这个等待的时间是相互独立的，有可能第一个人还在等，第二个人已经拿到货回来了。

参考技术A 在任何并发性应用程序中，异步事件处理都至关重要。事件来源可能是不同的计算任务、I/O 操作或与外部系统的交互。无论来源是什么，应用程序代码都必须跟踪事件，协调为响应事件而采取的操作。Java 应用程序可采用两种基本的异步事件处理方法：该应用程序有一个协调线程等待事件，然后采取操作，或者事件可在完成时直接执行某项操作（通常采取执行应用程序所提供的代码的方式）。让线程等待事件的方法被称为阻塞 方法。让事件执行操作、线程无需显式等待事件的方法被称为非阻塞 方法。
旧的 java.util.concurrent.Future 类提供了一种简单方式来处理预期的事件完成，但仅通过轮询或等待完成的方法。Java 8 中添加的 java.util.concurrent.CompletableFuture 类扩展了此功能，添加了大量合成或处理事件的方法。具体地讲，CompletableFuture 提供了在事件完成时执行应用程序代码的标准技术，包括各种组合任务（由 future 表示）的方式。这种组合技术使得编写非阻塞代码来处理事件变得很容易（或者至少比过去容易）。

Java Future源码分析

JDK future框架，提供了一种异步编程模式，基于线程池的。将任务runnable/callable提交到线程池executor，返回一个Future对象。通过future.get()获取执行结果，这里提交到线程池，后面的操作不会阻塞。future.get()获取结果会阻塞，其实也是用多线线程执行任务。

future.get()这里会阻塞，google的guava提供了一个calllback解决办法，这也是我准备看的

下面是一个future的demo

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * @Author: <[email protected]>
 * @Description: jdk future demo
 * @Date: Created in : 2018/11/30 6:01 PM
 **/
public class TestJdkFuture {

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    testJdkFuture();
  }


  public static void testJdkFuture() throws ExecutionException, InterruptedException {
    Callable<Integer> callable = () ->{
      System.out.println("callable do some compute");
      return 1;
    } ;
    Runnable runnable = () -> {
      System.out.println("runable do some compute");
    };
    ExecutorService executorService = Executors.newCachedThreadPool();
    Future<Integer> future = executorService.submit(callable);
    Future runableFuture = executorService.submit(runnable);
    Object runableRes = runableFuture.get();
    int res = future.get();
    executorService.shutdown();
    System.out.println("callable res: " + res);
    System.out.println("runnable res: " + runableRes);

  }
}

demo里面提交了一个Callable和一个Runnable到线程池，通过future获取计算结果

executorService.submit(callable)源码

public <T> Future<T> submit(Callable<T> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<T> ftask = newTaskFor(task);
        execute(ftask);
        return ftask;
    }

这里主要是封装了一个RunnableFuture和提交任务到线程池

我们看下RunnableFuture里面的run方法，因为线程池执行任务，是执行run方法。看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);
        }
    }

首先，设置runner为线程池中的当前线程，后面执行call()方法，计算出结果，set(result)。跟进set(result)

/**
     * Sets the result of this future to the given value unless
     * this future has already been set or has been cancelled.
     *
     * <p>This method is invoked internally by the {@link #run} method
     * upon successful completion of the computation.
     *
     * @param v the value
     */
    protected void set(V v) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
        }
    }

这里用了一个cas设置FutureTask的state字段为COMPLETING，完成中的一个状态。接着设置outcom为计算结果，我们跟进finishCompletion()

/**
     * Removes and signals all waiting threads, invokes done(), and
     * nulls out callable.
     */
    private void finishCompletion() {
        // assert state > COMPLETING;
        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
    }

WaitNode主要是一个阻塞线程链表，即调用future.get()方法的线程链表。这里主要作用是注意唤醒这些线程，通过LockSupport.unpark(t)唤醒。这里用阻塞线程链表，主要是考虑到可能有多个线程会调用future.get()阻塞

ok,到这里执行任务，把计算结果放到future中，并唤醒阻塞线程已经理清楚了

我们再来看下，future.get()是如何实现阻塞，和获取到计算结果的

进入future.get()

/**
     * 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;

Future.java是一个接口，这里看注释可以看出，会阻塞等待结果计算完成。我们看下一个实现FutureTask.java的get()方法

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

awaitDone(false, 0l)主要是阻塞线程，进入方法

/**
     * Awaits completion or aborts on interrupt or timeout.
     *
     * @param timed true if use timed waits
     * @param nanos time to wait, if timed
     * @return state upon completion
     */
    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);
        }
    }

这里，主要是阻塞线程，把当前线程放到阻塞线程链表中，通过LockSupport.park(this)阻塞当前线程，等待线程池里面的线程唤醒。唤醒之后，回到get()方法，看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);
    }

这里主要是返回计算结果给阻塞线程

到这里，基本理清楚了，future的阻塞实现，以及获取计算结果的步骤。

future框架 = 线程池 + futrue