最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

JUC中Future及CompletableFuture的用法與說明

 更新時間：2026年04月21日 09:54:35   作者：kkkwang0o0  
本文介紹了Future接口及其實現(xiàn)類FutureTask的使用方法,指出了其阻塞和CPU空轉(zhuǎn)問題,為了解決這些問題,介紹了CompletableFuture的異步任務(wù)編排,可以自動回調(diào)異步任務(wù)結(jié)束時的方法,解決了阻塞和輪詢問題,同時提供了鏈式調(diào)用、函數(shù)式編程等等提高編程效率

Future 異步回調(diào)（不推薦）

new Thread時傳入FutureTask對象（構(gòu)造時傳入Callable任務(wù)對象），調(diào)用start啟動

創(chuàng)建線程的方式

1.直接new Thread()對象，重寫run方法，調(diào)用start啟動

2.new Thread時傳入Runnable任務(wù)對象（重寫run方法），調(diào)用start啟動

3.new Thread時傳入FutureTask對象（構(gòu)造時傳入Callable任務(wù)對象），調(diào)用start啟動

/**
 * 創(chuàng)建線程的三種方式
 */
public class CreateThreadDemo {
    public static void main(String[] args) throws Exception {
        // 1.重寫Thread的run方法
        Thread t1 = new Thread(() -> {
            System.out.println("第一種創(chuàng)建線程方式，重寫Thread的run方法");
        },"t1");
        // 2.重寫Runnable的run方法
        RunTask runTask = new RunTask();
        Thread t2 = new Thread(runTask, "t2");
        // 3.傳入FutureTask對象，重寫Callable的call方法（異步帶返回值）
        CallTask callTask = new CallTask();
        FutureTask<String> futureTask = new FutureTask<>(callTask);
        Thread t3 = new Thread(futureTask, "t3");
        t1.start();
        t2.start();
        t3.start();
        System.out.println("異步執(zhí)行結(jié)果" + futureTask.get() + " " + System.currentTimeMillis());
    }
}
/**
 * Runnable 任務(wù)
 */
class RunTask implements Runnable {
    @Override
    public void run() {
        System.out.println("runnable running " + System.currentTimeMillis());
    }
}
/**
 * Callable 任務(wù)
 */
class CallTask implements Callable<String> {
    @Override
    public String call() throws Exception {
        System.out.println("callable running " + System.currentTimeMillis());
        return "hello";
    }
}

概述

Fucture接口（FutureTask實現(xiàn)類）定義了操作異步任務(wù)執(zhí)行的一些方法，如獲取異步任務(wù)的執(zhí)行結(jié)果、取消任務(wù)的執(zhí)行、判斷任務(wù)是否被取消、判斷任務(wù)是否完畢等。

Future接口可以為主線程開一個分支任務(wù)，專門為主線程處理耗時費力的復(fù)雜業(yè)務(wù)。

Future提供了一種異步并行計算的功能：如果主線程需要執(zhí)行一個很耗時的計算任務(wù)，我們就可以通過future把這個任務(wù)放到異步線程中執(zhí)行，主線程繼續(xù)處理其他任務(wù)或者先行結(jié)束，再通過Future獲取計算結(jié)果。

Future接口 --- 源碼

public interface Future<V> {
    boolean cancel(boolean mayInterruptIfRunning);
    boolean isCancelled();
    boolean isDone();
    V get() throws InterruptedException, ExecutionException;
    V get(long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException;
}

FutureTask（Future實現(xiàn)類）

Thread類構(gòu)造方法：

可以看出Thread構(gòu)造方法接收的任務(wù)對象只有Runnable接口，為什么還能接收FutureTask任務(wù)對象呢？

因為FutureTask實現(xiàn)了RunableFuture接口，而RunnableFuture接口繼承了Runnable接口和Future接口，所以FutureTask也是Runnable對象，而FutureTask可以接收Callable任務(wù)對象是因為構(gòu)造方法中提供了接收Callable對象的構(gòu)造方法。

Future編碼優(yōu)缺點

  • 優(yōu)點：future+線程池可以實現(xiàn)多線程任務(wù)配合，能顯著提高程序的執(zhí)行效率
  • 缺點：get()阻塞、isDone()輪詢導(dǎo)致CPU空轉(zhuǎn)
  • 阻塞情況： 調(diào)用FutureTask.get方法時線程會阻塞等待異步結(jié)果的獲取

代碼：

FutureTask<String> futureTask = new FutureTask<String>(() -> {
    System.out.println("開始執(zhí)行......" + now().getSecond());
    // 休眠
    try {
        TimeUnit.SECONDS.sleep(5);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    System.out.println("結(jié)束......");
    return "FutureTask End";
});
Thread t = new Thread(futureTask);
t.start();
System.out.println("子線程執(zhí)行結(jié)果：" + futureTask.get());
System.out.println("--------------------------------");
System.out.println("主線程 -------- 執(zhí)行..... " + now().getSecond());
System.out.println("--------------------------------");

運行結(jié)果：

可以通過設(shè)置：

System.out.println("子線程執(zhí)行結(jié)果：" + futureTask.get(2, TimeUnit.SECONDS));

暴力終止程序運行或拋出異常

isDone()輪詢導(dǎo)致CPU空轉(zhuǎn)：

輪詢的方式會消耗無謂的CPU資源，而且也不見得能及時獲得到計算結(jié)果，如果想要異步獲取結(jié)果，通常都會以輪詢的方式去獲取結(jié)果，盡量不要阻塞。

while (true) {
    if (futureTask.isDone()) {
        System.out.println("子線程執(zhí)行結(jié)果：" + futureTask.get() + now().getSecond());
        break;
    }else {
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("暫停2s");
    }
}

結(jié)論：Future對于結(jié)果的獲取不是很友好，只能通過阻塞或者輪詢的方式得到任務(wù)的結(jié)果。

由此引入CompletableFuture，規(guī)避以上的缺點

需要說明的是對于簡單的業(yè)務(wù)使用Future就可以了

CompletableFuture（異步任務(wù)編排）

CompletableFuture可以代表一個明確完成的Future，也可能代表一個完成階段(CompletionStage),它支持在計算完成以后觸發(fā)一些函數(shù)或執(zhí)行某些動作。

CompletableFuture可以解決的問題？

多個任務(wù)前后依賴組合處理：想將多個異步任務(wù)的計算結(jié)果組合起來，后一個異步任務(wù)的計算結(jié)果需要前一個異步任務(wù)的值；將兩個或多個異步計算合成一個異步計算，這幾個異步計算互相獨立，同時后面這個又依賴前一個計算結(jié)果

對計算速度選最快：當Future集合中某個任務(wù)最快結(jié)束時，返回結(jié)果，返回第一名的處理結(jié)果。。。。

CompletableFuture對比Future的改進

Future問題：get()阻塞、isDone()輪詢CPU空轉(zhuǎn)

對于真正的異步處理我們希望是可以通過傳入回調(diào)函數(shù)，在Future結(jié)束時自動調(diào)用該回調(diào)函數(shù)，這樣，我們就不用等待結(jié)果。

CompletableFuture提供一種觀察者模式類似的機制，可以讓任務(wù)執(zhí)行完成后通知監(jiān)聽的一方。

怎么創(chuàng)建CompletableFuture對象？

創(chuàng)建

一般不建議直接new CompletableFuture

可以采用以下的方式創(chuàng)建CompletableFuture對象：

1.CompletableFuture.runAsync無返回值（指定線程池會用指定的線程池，沒有就會使用默認的線程池）

2. CompletableFuture.supplyAsync有返回值（指定線程池會用指定的線程池，沒有就會使用默認的線程池）

怎么解決Future的阻塞和輪詢問題？

從java8開始引入了CompletableFuture，它是Future的功能增強版，減少阻塞和輪詢，可以傳入回調(diào)對象，當異步任務(wù)完成或者發(fā)生異常時，自動調(diào)用回調(diào)對象的回調(diào)方法。

默認線程池創(chuàng)建的是守護線程，自定義線程池是用戶線程

public static void main(String[] args) throws ExecutionException, InterruptedException {
    ExecutorService threadPool = Executors.newFixedThreadPool(5);
    // 有返回值
    supplyFuture(threadPool);
    System.out.println("主線程 running " + now());
    threadPool.shutdown();
}
private static void supplyFuture(ExecutorService threadPool) {
    CompletableFuture<Integer> supplyFuture = CompletableFuture.supplyAsync(() -> {
        System.out.println("開始執(zhí)行...." +
                (Thread.currentThread().isDaemon() ? "守護線程" : "用戶線程") +
                " " + now());
        int random = ThreadLocalRandom.current().nextInt();
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return random;
    }, threadPool).whenComplete((v, e) -> {
        if (e == null) {
            // 沒有異常
            System.out.println("隨機數(shù)結(jié)果： " + v);
        }
    }).exceptionally((e) -> {
        e.printStackTrace();
        System.out.println("異常情況：" + e.getCause());
        return null;
    });
}

運行結(jié)果：

已經(jīng)解決了Future編碼出現(xiàn)的get阻塞問題。

上述代碼塊邏輯解析：

supplyAsync()中是任務(wù)（默認線程池創(chuàng)建的守護線程），如果這個任務(wù)成功會走到whenComplete代碼塊中，不成功會走到exceptionally代碼塊中

注意：默認線程池創(chuàng)建的是守護線程，自定義線程池是用戶線程，守護線程會隨著用戶線程的結(jié)束而結(jié)束，所以會導(dǎo)致主線程執(zhí)行完了然后還沒打印出隨機數(shù)線程池就關(guān)閉了，就是以下輸出的情況

怎么獲得異步結(jié)果？

通過get或者join獲得結(jié)果（區(qū)別在于join在編譯期間不會作檢查性異常的處理，拋不拋異常都可以）

CompletableFuture的優(yōu)點

  1. 異步任務(wù)結(jié)束時，會自動回調(diào)某個對象的方法；
  2. 主線程設(shè)置好回調(diào)后，不再關(guān)心異步任務(wù)的執(zhí)行，異步任務(wù)之間可以順序執(zhí)行；
  3. 異步任務(wù)出錯時，會自動回調(diào)某個對象的方法：

函數(shù)式編程

函數(shù)式編程：Lambda表達式+Stream流式編程+Chain鏈式調(diào)用+Java8函數(shù)式編程

如Runnable、Function、Consumer --- BIConsumer、Supplier

常用函數(shù)式接口

函數(shù)式接口名稱

方法名稱

參數(shù)

返回值類型

Consumer<T>

accept

T t

void

Supplier<T>

get

無

T

Function<T, R>

apply

T t

R

Predicate<T>

test

T t

boolean

BiConsumer<T, U>

accept

T t, U u

void

BiFunction<T, U, R>

apply

T t, U u

R

BiPredicate<T, U>

test

T t, U u

boolean

UnaryOperator<T>

apply

T t

T

BinaryOperator<T>

apply

T t1, T t2

T

IntConsumer

accept

int value

void

IntSupplier

getAsInt

無

int

IntFunction<R>

apply

int value

R

IntPredicate

test

int value

boolean

鏈式語法

public class ChainDemo {
    public static void main(String[] args) {
        Student student = new Student();
        student.setId(1).setName("karry").setMajor("cs");
    }
}
@Data
@AllArgsConstructor
@NoArgsConstructor
@Accessors(chain = true)
class Student{
    private Integer id;
    private String name;
    private String major;
}

案例精講-電商網(wǎng)站的比價需求

從“功能”到“性能”

/**
 * 電商網(wǎng)站的比價需求
 */
public class CompletableMallDemo {
    private static final List<Mall> list = new ArrayList<>();
    static {
        for (int i = 0; i <= 1000; i ++) {
            list.add(new Mall("book" + i));
        }
    }
    /**
     * 查詢價格 同步處理
     * @param list 平臺列表
     * @param bookName 書籍名稱
     * @return 查詢所需時間
     */
    public static Long getPrice(List<Mall> list, String bookName) {
        long start = System.currentTimeMillis();
        List<String> prices = list.stream()
        .map((mall) -> String.format(bookName + " in %s price is %.2f",
                                     mall.getName(), mall.calcPrice(bookName)))
        .collect(Collectors.toList());
        long end = System.currentTimeMillis();
        System.out.println("用時:  " + end);
        prices.forEach(System.out::println);
        return end - start;
    }
    /**
     * 查詢價格 異步處理
     * @param list 平臺
     * @param bookName 書籍名稱
     * @return 查詢所需
     */
    public static Long getPriceByCompletableFuture(List<Mall> list, String bookName) {
        long start = System.currentTimeMillis();
        ExecutorService threadPool = Executors.newFixedThreadPool(10);
        List<String> prices = list.stream()
        // 把 mall 映射到 CompletableFuture對象
        .map(mall -> CompletableFuture.supplyAsync(() ->
                                                   String.format(bookName + " in %s price is %.2f",
                                                                 mall.getName(), mall.calcPrice(bookName)), threadPool)
            ).collect(Collectors.toList()).stream()
        // CompletableFuture對象 映射到 CompletableFuture.join() [String對象]
        .map(CompletableFuture::join).collect(Collectors.toList());
        threadPool.shutdown();
        long end = System.currentTimeMillis();
        System.out.println("用時:  " + end);
        prices.forEach(System.out::println);
        return end - start;
    }
    public static void main(String[] args) {
        System.out.println("差值：" +
                           (getPrice(list, "mysql") - getPriceByCompletableFuture(list, "mysql")));
    }
}
@AllArgsConstructor
@NoArgsConstructor
@Data
class Mall {
    /**
     * 電商網(wǎng)站名稱
     */
    private String name;
    /**
     * 模擬查詢價格
     * @param bookName 書籍名稱
     * @return double 價格
     */
    public double calcPrice(String bookName) {
        return ThreadLocalRandom.current().nextDouble(20, 100) + bookName.charAt(0);
    }
}

源碼分析

public class CompletableFuture<T> implements Future<T>, CompletionStage<T>  {
}

CompletableFuture實現(xiàn)了Future接口，還拓展了Future不具備的CompletionStage接口

CompletionStage（完成階段）

CompletionStage代表異步計算過程中的某一個階段，一個階段完成后可能會觸發(fā)另外一個階段。

一個階段的計算執(zhí)行可以時一個Fuction，Consumer或者Runnable。比如：stage.thenApply().thenAccept().thenRun()

CompletableFuture API

不帶Async和帶Async的API區(qū)別：

對于CompletionStage，每個任務(wù)階段帶Async的任務(wù)可以設(shè)定線程池，不設(shè)定就會使用默認線程池, 而不帶Async的任務(wù)階段會使用CompletableFuture設(shè)定的線程池。

thenRun和thenRunAsync的區(qū)別：

  • thenRun使用的supplyAsync或者runAsync傳入的線程池（不傳入則使用默認線程池---守護線程）
  • thenRunAsync使用的是自己api傳入的線程池，不傳入則使用默認線程池（守護線程）
/**
 * thenRunAsync方法
 * @param threadPool
 */
public static void testRunAsyncMethod(ExecutorService threadPool) {
CompletableFuture.runAsync(() -> {
    System.out.println("step 1 " + Thread.currentThread().getName());
}, threadPool).thenRunAsync(() -> {
    System.out.println("step 2 " + Thread.currentThread().getName());
}).thenRunAsync(() -> {
    System.out.println("step 3 " + Thread.currentThread().getName());
}, threadPool).thenRunAsync(() -> {
    System.out.println("step 4 " + Thread.currentThread().getName());
}).thenRunAsync(() -> {
    System.out.println("step 5 " + Thread.currentThread().getName());
}).join();
}

方法

觸發(fā)時機

輸入?yún)?shù)

返回值類型

核心作用

thenRun

前序任務(wù)正常完成后

無（Runnable

）

CompletableFuture<Void>

執(zhí)行無輸入、無輸出的后續(xù)操作

thenAccept

前序任務(wù)正常完成后

前序任務(wù)的結(jié)果（Consumer

）

CompletableFuture<Void>

消費前序任務(wù)的結(jié)果，無新輸出

thenApply

前序任務(wù)正常完成后

前序任務(wù)的結(jié)果（Function

）

CompletableFuture<U>

基于前序結(jié)果計算新結(jié)果，有新輸出

handle

前序任務(wù)完成（無論成?。?/p>

前序結(jié)果 + 異常（BiFunction

）

CompletableFuture<U>

處理前序任務(wù)的結(jié)果或異常，計算新結(jié)果

1. 獲得結(jié)果和觸發(fā)計算

獲得結(jié)果：

  • get() : 不見不散
  • get(long, TimUnit): 過時不候
  • join() : 功能和get類似，但是在編譯期間不拋出受檢查異常
  • getNow(valueIfAbsent): 立即獲得當前結(jié)果，為空則返回valueIfAbsent的值
public T getNow(T valueIfAbsent) {
Object r;
return ((r = result) == null) ? valueIfAbsent : reportJoin(r);
}

觸發(fā)計算：

complete(T value) ：是否打斷get方法立即返回括號值（返回括號值，為true；不返回為false）

public boolean complete(T value) {
    boolean triggered = completeValue(value);
    postComplete();
    return triggered;
}

eg:

public static void main(String[] args) throws InterruptedException {
    ExecutorService threadPool = Executors.newFixedThreadPool(5);
    testCompleteMethodByTrue(threadPool);
    testCompleteMethodByFalse(threadPool);
    threadPool.shutdown();
}
public static void testCompleteMethodByFalse(ExecutorService threadPool) throws InterruptedException {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "get/join返回的結(jié)果";
    }, threadPool);
    TimeUnit.SECONDS.sleep(4);
    boolean flag = future.complete("設(shè)定的值");
    System.out.println("complete方法返回值為 " + flag + "，4s左右獲得的值為 " + future.join());
}
public static void testCompleteMethodByTrue(ExecutorService threadPool) throws InterruptedException {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "get/join返回的結(jié)果";
    }, threadPool);
    TimeUnit.SECONDS.sleep(2);
    boolean flag = future.complete("設(shè)定的值");
    System.out.println("complete方法返回值為 " + flag + "，2s左右獲得的值為 " + future.join());
}

運行結(jié)果：

2. 對計算結(jié)果進行處理

2.1 thenApply: 計算結(jié)果存在依賴關(guān)系，串行化

/**
 * thenApply
 * @param threadPool
 */
public static void testApplyMethod(ExecutorService threadPool) {
StringBuffer str = new StringBuffer();
CompletableFuture<StringBuffer> future = CompletableFuture.supplyAsync(() -> {
    return str.append("a");
}, threadPool).thenApply(f -> {
    return str.append("b");
}).thenApply(f -> {
    return str.append("c");
}).whenComplete((v, e) -> {
    if (v != null) {
        System.out.println("apply處理后的結(jié)果為 " + v);
    }
}).exceptionally(e -> {
    e.printStackTrace();
    return null;
});
}

2.2 handle： 計算結(jié)果存在依賴關(guān)系，串行化（可以帶著）

/**
 * handle方法
 * @param threadPool
 */
public static void testHandleMethod(ExecutorService threadPool) {
    StringBuffer str = new StringBuffer();
    CompletableFuture<StringBuffer> future = CompletableFuture.supplyAsync(() -> {
        return str.append("a");
    }, threadPool).handle((f, e) -> {
        if (e == null) {
            return f.append("b");
        }else {
            e.printStackTrace();
            return null;
        }
    }).handle((f, e) -> {
        if (e == null) {
            return f.append("c");
        }else {
            e.printStackTrace();
            return null;
        }
    });
    System.out.println("handle處理后的結(jié)果為 " + future.join());
}

3.對計算結(jié)果進行消費

接收任務(wù)的處理結(jié)果，消費處理，無返回結(jié)果

thenAccept:

public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
    return uniAcceptStage(null, action);
}
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
    return uniAcceptStage(asyncPool, action);
}
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
                                               Executor executor) {
    return uniAcceptStage(screenExecutor(executor), action);
}

eg:

/**
 * thenAccept方法（消費型）
 * @param threadPool
 */
public static void testAcceptMethod(ExecutorService threadPool) {
    StringBuffer str = new StringBuffer();
    CompletableFuture.supplyAsync(() -> str.append("abc"), threadPool)
    .thenAccept(r -> System.out.println("accept直接消費" + r)).join();
}

輸出：

thenRun:

任務(wù)A執(zhí)行完執(zhí)行B，并且B不需要A的結(jié)果

/**
 * thenRun方法
 * @param threadPool
 */
public static void testRunMethod(ExecutorService threadPool) {
CompletableFuture.runAsync(() -> {
    System.out.println("step 1");
}, threadPool).thenRun(() -> {
    System.out.println("step 2");
}).thenRun(() -> {
    System.out.println("step 3");
});
}
4.對計算速度進行選用

applyToEither:

對兩個future對象選用速度較快的那一個結(jié)果

eg:

/**
 * applyToEither 方法
 * @param threadPool
 */
public static void testApplyToEitherMethod(ExecutorService threadPool) {
for (int i = 2; i <= 5; i ++) {
    CompletableFuture<String> future = getPlayFuture(i - 1, threadPool).applyToEither(getPlayFuture(i, threadPool), f -> f + " is winner");
    System.out.println(future.join());
}
}
private static CompletableFuture<String> getPlayFuture(int num, ExecutorService threadPool) {
    return CompletableFuture.supplyAsync(() -> {
        try {
            TimeUnit.MILLISECONDS.sleep(num * 100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "play" + num;
    });
}

5.對計算結(jié)果進行合并

thenCombine:

兩個CompletionStage任務(wù)都完成后，最終能把兩個任務(wù)的結(jié)果一起提交給thenCombine來處理；先完成的先等著，等待其它分支任務(wù)。

eg：

/**
 * thenCombine 方法
 * @param threadPool
 */
public static void testCombineMethod(ExecutorService threadPool) {
    CompletableFuture<Integer> future1 = getBranchFuture(1, threadPool);
    CompletableFuture<Integer> future2 = getBranchFuture(2, threadPool);
    CompletableFuture<Integer> combineFuture = future2.thenCombine(future1, Integer::sum);
    System.out.println("計算結(jié)果為 " + combineFuture.join());
}
private static CompletableFuture<Integer> getBranchFuture(int num, ExecutorService threadPool) {
    return CompletableFuture.supplyAsync(() -> {
        try {
            TimeUnit.MILLISECONDS.sleep(num * 100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return num * 10;
    });
}

輸出：計算結(jié)果為 30

總結(jié)

以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。

相關(guān)文章

最新評論

永宁县| 皮山县| 师宗县| 清涧县| 卢龙县| 商南县| 通许县| 宁强县| 新竹县| 康平县| 常州市| 扎赉特旗| 都昌县| 贵港市| 汉川市| 湖北省| 礼泉县| 临夏市| 平湖市| 阿鲁科尔沁旗| 突泉县| 银川市| 永康市| 灵武市| 鹰潭市| 二手房| 河西区| 蒙阴县| 普兰县| 通州区| 平潭县| 玛多县| 三门峡市| 福安市| 京山县| 阳曲县| 昭觉县| 盐津县| 奎屯市| 赣榆县| 同江市|