關(guān)于Java?中?Future?的?get?方法超時(shí)問(wèn)題

更新時(shí)間：2022年06月15日 08:56:53 作者：time-flies

這篇文章主要介紹了Java?中?Future?的?get?方法超時(shí),最常見(jiàn)的理解就是，“超時(shí)以后，當(dāng)前線程繼續(xù)執(zhí)行，線程池里的對(duì)應(yīng)線程中斷”，真的是這樣嗎？本文給大家詳細(xì)介紹，需要的朋友參考下吧

一、背景

很多 Java 工程師在準(zhǔn)備面試時(shí)，會(huì)刷很多八股文，線程和線程池這一塊通常會(huì)準(zhǔn)備線程的狀態(tài)、線程的創(chuàng)建方式，Executors 里面的一些工廠方法和為什么不推薦使用這些工廠方法，ThreadPoolExecutor 構(gòu)造方法的一些參數(shù)和執(zhí)行過(guò)程等。

工作中，很多人會(huì)使用線程池的 submit 方法獲取 Future 類型的返回值，然后使用 java.util.concurrent.Future#get(long, java.util.concurrent.TimeUnit) 實(shí)現(xiàn)“最多等多久”的效果。

但很多人對(duì)此的理解只停留在表面上，稍微問(wèn)深一點(diǎn)點(diǎn)可能就懵逼了。

比如，java.util.concurrent.Future#get(long, java.util.concurrent.TimeUnit) 超時(shí)之后，當(dāng)前線程會(huì)怎樣？線程池里執(zhí)行對(duì)應(yīng)任務(wù)的線程會(huì)有怎樣的表現(xiàn)？

如果你對(duì)這個(gè)問(wèn)題沒(méi)有很大的把握，說(shuō)明你掌握的還不夠扎實(shí)。

最常見(jiàn)的理解就是，“超時(shí)以后，當(dāng)前線程繼續(xù)執(zhí)行，線程池里的對(duì)應(yīng)線程中斷”，真的是這樣嗎？

二、模擬

2.1 常見(jiàn)寫法

下面給出一個(gè)簡(jiǎn)單的模擬案例：

package basic.thread;
import java.util.concurrent.*;
public class FutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        Future<?> future = executorService.submit(() -> {
            try {
                demo();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        String threadName = Thread.currentThread().getName();
        System.out.println(threadName + "獲取的結(jié)果 -- start");
        Object result = future.get(100, TimeUnit.MILLISECONDS);
        System.out.println(threadName + "獲取的結(jié)果 -- end :" + result);
    }
    private static String demo() throws InterruptedException {
        String threadName = Thread.currentThread().getName();
        System.out.println(threadName + ",執(zhí)行 demo -- start");
        TimeUnit.SECONDS.sleep(1);
        System.out.println(threadName + ",執(zhí)行 demo -- end");
        return "test";
    }
}

輸出結(jié)果：

main獲取的結(jié)果 -- start
pool-1-thread-1,執(zhí)行 demo -- start
Exception in thread "main" java.util.concurrent.TimeoutException
at java.util.concurrent.FutureTask.get(FutureTask.java:205)
at basic.thread.FutureDemo.main(FutureDemo.java:20)
pool-1-thread-1,執(zhí)行 demo -- end

我們可以發(fā)現(xiàn)：當(dāng)前線程會(huì)因?yàn)槭盏?TimeoutException 而被中斷，線程池里對(duì)應(yīng)的線程“卻”繼續(xù)執(zhí)行完畢。

2.2 嘗試取消

我們嘗試對(duì)未完成的線程進(jìn)行取消，發(fā)現(xiàn) Future#cancel 有個(gè) boolean 類型的參數(shù)。

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

看源碼注釋我們可以知道：

當(dāng)設(shè)置為 true 時(shí)，正在執(zhí)行的任務(wù)將被中斷（interrupted）；

當(dāng)設(shè)置為 false 時(shí)，如果任務(wù)正在執(zhí)行中，那么仍然允許任務(wù)執(zhí)行完成。

2.2.1 cancel(false)

此時(shí)，為了不讓主線程因?yàn)槌瑫r(shí)異常被中斷，我們 try-catch 包起來(lái)。

package basic.thread;
import org.junit.platform.commons.util.ExceptionUtils;
import java.util.concurrent.*;
public class FutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        Future<?> future = executorService.submit(() -> {
            try {
                demo();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        String threadName = Thread.currentThread().getName();
        System.out.println(System.currentTimeMillis() + "," + threadName + "獲取的結(jié)果 -- start");
        try {
            Object result = future.get(100, TimeUnit.MILLISECONDS);
            System.out.println(System.currentTimeMillis() + "," + threadName + "獲取的結(jié)果 -- end :" + result);
        } catch (Exception e) {
            System.out.println(System.currentTimeMillis() + "," + threadName + "獲取的結(jié)果異常:" + ExceptionUtils.readStackTrace(e));
        }
        future.cancel(false);
        System.out.println(System.currentTimeMillis() + "," + threadName + "獲取的結(jié)果 -- cancel");
    }
    private static String demo() throws InterruptedException {
        String threadName = Thread.currentThread().getName();
        System.out.println(System.currentTimeMillis() + "," + threadName + ",執(zhí)行 demo -- start");
        TimeUnit.SECONDS.sleep(1);
        System.out.println(System.currentTimeMillis() + "," + threadName + ",執(zhí)行 demo -- end");
        return "test";
    }
}

結(jié)果：

1653751759233,main獲取的結(jié)果 -- start
1653751759233,pool-1-thread-1,執(zhí)行 demo -- start
1653751759343,main獲取的結(jié)果異常:java.util.concurrent.TimeoutException
at java.util.concurrent.FutureTask.get(FutureTask.java:205)
at basic.thread.FutureDemo.main(FutureDemo.java:23)

1653751759351,main獲取的結(jié)果 -- cancel
1653751760263,pool-1-thread-1,執(zhí)行 demo -- end

我們發(fā)現(xiàn)，線程池里的對(duì)應(yīng)線程在 cancel(false) 時(shí)，如果已經(jīng)正在執(zhí)行，則會(huì)繼續(xù)執(zhí)行完成。

2.2.2 cancel(true)

package basic.thread;
import org.junit.platform.commons.util.ExceptionUtils;
import java.util.concurrent.*;
public class FutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        Future<?> future = executorService.submit(() -> {
            try {
                demo();
            } catch (InterruptedException e) {
                System.out.println(System.currentTimeMillis() + "," + Thread.currentThread().getName() + ", Interrupted:" + ExceptionUtils.readStackTrace(e));
                throw new RuntimeException(e);
            }
        });
        String threadName = Thread.currentThread().getName();
        System.out.println(System.currentTimeMillis() + "," + threadName + "獲取的結(jié)果 -- start");
        try {
            Object result = future.get(100, TimeUnit.MILLISECONDS);
            System.out.println(System.currentTimeMillis() + "," + threadName + "獲取的結(jié)果 -- end :" + result);
        } catch (Exception e) {
            System.out.println(System.currentTimeMillis() + "," + threadName + "獲取的結(jié)果異常:" + ExceptionUtils.readStackTrace(e));
        }
        future.cancel(true);
        System.out.println(System.currentTimeMillis() + "," + threadName + "獲取的結(jié)果 -- cancel");
    }
    private static String demo() throws InterruptedException {
        String threadName = Thread.currentThread().getName();
        System.out.println(System.currentTimeMillis() + "," + threadName + ",執(zhí)行 demo -- start");
        TimeUnit.SECONDS.sleep(1);
        System.out.println(System.currentTimeMillis() + "," + threadName + ",執(zhí)行 demo -- end");
        return "test";
    }
}

執(zhí)行結(jié)果：

1653752011246,main獲取的結(jié)果 -- start
1653752011246,pool-1-thread-1,執(zhí)行 demo -- start
1653752011347,main獲取的結(jié)果異常:java.util.concurrent.TimeoutException
   at java.util.concurrent.FutureTask.get(FutureTask.java:205)
   at basic.thread.FutureDemo.main(FutureDemo.java:24)

1653752011363,pool-1-thread-1, Interrupted:java.lang.InterruptedException: sleep interrupted
   at java.lang.Thread.sleep(Native Method)
   at java.lang.Thread.sleep(Thread.java:340)
   at java.util.concurrent.TimeUnit.sleep(TimeUnit.java:386)
   at basic.thread.FutureDemo.demo(FutureDemo.java:36)
   at basic.thread.FutureDemo.lambda$main$0(FutureDemo.java:14)
   at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
   at java.util.concurrent.FutureTask.run(FutureTask.java:266)
   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)

1653752011366,main獲取的結(jié)果 -- cancel

可以看出，此時(shí)，如果目標(biāo)線程未執(zhí)行完，那么會(huì)收到 InterruptedException ，被中斷。

當(dāng)然，如果此時(shí)不希望目標(biāo)線程被中斷，可以使用 try-catch 包住，再執(zhí)行其他邏輯。

package basic.thread;
import org.junit.platform.commons.util.ExceptionUtils;
import java.util.concurrent.*;
public class FutureDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        Future<?> future = executorService.submit(() -> {
            demo();
        });
        String threadName = Thread.currentThread().getName();
        System.out.println(System.currentTimeMillis() + "," + threadName + "獲取的結(jié)果 -- start");
        try {
            Object result = future.get(100, TimeUnit.MILLISECONDS);
            System.out.println(System.currentTimeMillis() + "," + threadName + "獲取的結(jié)果 -- end :" + result);
        } catch (Exception e) {
            System.out.println(System.currentTimeMillis() + "," + threadName + "獲取的結(jié)果異常:" + ExceptionUtils.readStackTrace(e));
        }
        future.cancel(true);
        System.out.println(System.currentTimeMillis() + "," + threadName + "獲取的結(jié)果 -- cancel");
    }
    private static String demo() {
        String threadName = Thread.currentThread().getName();
        System.out.println(System.currentTimeMillis() + "," + threadName + ",執(zhí)行 demo -- start");
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            System.out.println(System.currentTimeMillis() + "," + threadName + ",執(zhí)行 demo 被中斷，自動(dòng)降級(jí)");
        }
        System.out.println(System.currentTimeMillis() + "," + threadName + ",執(zhí)行 demo -- end");
        return "test";
    }
}

執(zhí)行結(jié)果：

1653752219803,main獲取的結(jié)果 -- start
1653752219803,pool-1-thread-1,執(zhí)行 demo -- start
1653752219908,main獲取的結(jié)果異常:java.util.concurrent.TimeoutException
at java.util.concurrent.FutureTask.get(FutureTask.java:205)
at basic.thread.FutureDemo.main(FutureDemo.java:19)

1653752219913,main獲取的結(jié)果 -- cancel
1653752219914,pool-1-thread-1,執(zhí)行 demo 被中斷，自動(dòng)降級(jí)
1653752219914,pool-1-thread-1,執(zhí)行 demo -- end

三、回歸源碼

我們直接看 java.util.concurrent.Future#get(long, java.util.concurrent.TimeUnit) 的源碼注釋，就可以清楚地知道各種情況的表現(xiàn)：

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

我們還可以選取幾個(gè)常見(jiàn)的實(shí)現(xiàn)類，查看下實(shí)現(xiàn)的基本思路：

java.util.concurrent.FutureTask#get(long, java.util.concurrent.TimeUnit)

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

java.util.concurrent.CompletableFuture#get(long, java.util.concurrent.TimeUnit)

    /**
     * Waits if necessary for at most the given time for this future
     * to complete, and then returns its result, if available.
     *
     * @param timeout the maximum time to wait
     * @param unit the time unit of the timeout argument
     * @return the result value
     * @throws CancellationException if this future was cancelled
     * @throws ExecutionException if this future completed exceptionally
     * @throws InterruptedException if the current thread was interrupted
     * while waiting
     * @throws TimeoutException if the wait timed out
     */
    public T get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
        Object r;
        long nanos = unit.toNanos(timeout);
        return reportGet((r = result) == null ? timedGet(nanos) : r);
    }

  /**
     * Returns raw result after waiting, or null if interrupted, or
     * throws TimeoutException on timeout.
     */
    private Object timedGet(long nanos) throws TimeoutException {
        if (Thread.interrupted())
            return null;
        if (nanos <= 0L)
            throw new TimeoutException();
        long d = System.nanoTime() + nanos;
        Signaller q = new Signaller(true, nanos, d == 0L ? 1L : d); // avoid 0
        boolean queued = false;
        Object r;
        // We intentionally don't spin here (as waitingGet does) because
        // the call to nanoTime() above acts much like a spin.
        while ((r = result) == null) {
            if (!queued)
                queued = tryPushStack(q);
            else if (q.interruptControl < 0 || q.nanos <= 0L) {
                q.thread = null;
                cleanStack();
                if (q.interruptControl < 0)
                    return null;
                throw new TimeoutException();
            }
            else if (q.thread != null && result == null) {
                try {
                    ForkJoinPool.managedBlock(q);
                } catch (InterruptedException ie) {
                    q.interruptControl = -1;
                }
            }
        }
        if (q.interruptControl < 0)
            r = null;
        q.thread = null;
        postComplete();
        return r;
    }

java.util.concurrent.Future#cancel 也一樣

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

java.util.concurrent.FutureTask#cancel

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

可以看到 mayInterruptIfRunning 為 true 時(shí)，會(huì)執(zhí)行 Thread#interrupt 方法

java.util.concurrent.CompletableFuture#cancel

    /**
     * If not already completed, completes this CompletableFuture with
     * a {@link CancellationException}. Dependent CompletableFutures
     * that have not already completed will also complete
     * exceptionally, with a {@link CompletionException} caused by
     * this {@code CancellationException}.
     *
     * @param mayInterruptIfRunning this value has no effect in this
     * implementation because interrupts are not used to control
     * processing.
     *
     * @return {@code true} if this task is now cancelled
     */
    public boolean cancel(boolean mayInterruptIfRunning) {
        boolean cancelled = (result == null) &&
            internalComplete(new AltResult(new CancellationException()));
        postComplete();
        return cancelled || isCancelled();
    }

通過(guò)注釋我們也發(fā)現(xiàn)，不同的實(shí)現(xiàn)類對(duì)參數(shù)的“效果”也有差異。

四、總結(jié)

我們學(xué)習(xí)時(shí)不應(yīng)該想當(dāng)然，不能紙上談兵，對(duì)于不太理解的地方，可以多看源碼注釋，多看源碼，多寫 DEMO 去模擬或調(diào)試。

到此這篇關(guān)于Java 中 Future 的 get 方法超時(shí)會(huì)怎樣的文章就介紹到這了,更多相關(guān)Java Future 的 get 超時(shí)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: