執行緒池怎么做到主動回呼

某些業務場景需要在一個異步操作完成后做一個回呼，如果使用Future有個問題，需要主動呼叫get然后再執行邏輯，但是異步邏輯不知道什么時候結束，所以推薦回呼的方式處理比較優雅，本文總結了三種方式可以實作此類需求，

1、CompletableFuture

Future可以明確地完成（設定其值和狀態），并且可以被用作CompletionStage ，支持有關的功能和它的完成時觸發動作，

當兩個或多個執行緒試圖complete ， completeExceptionally ，或cancel一個CompletableFuture，只有一個成功，

除了直接操作狀態和結果的這些和相關方法外，CompletableFuture還實作了介面CompletionStage ，具有以下策略：

• 為異步方法的依賴完成提供的操作可以由完成當前CompletableFuture的執行緒或完成方法的任何其他呼叫者執行，
• 所有不使用顯式Executor引數的異步方法都使用ForkJoinPool.commonPool()執行 （除非它不支持至少兩個并行級別，在這種情況下，使用新的執行緒）， 為了簡化監視，除錯和跟蹤，所有生成的異步任務都是標記介面CompletableFuture.AsynchronousCompletionTask的實體 ，
• 所有CompletionStage方法都是獨立于其他公共方法實作的，因此一個方法的行為不會受到子類中其他方法的覆寫的影響，

CompletableFuture還實施Future ，具有以下政策：

• 由于（不同于FutureTask ），這個類不能直接控制導致其完成的計算，所以取消被視為另一種形式的例外完成， 方法cancel具有相同的效果completeExceptionally(new CancellationException()) ， 方法isCompletedExceptionally()可用于確定CompletableFuture是否以任何特殊方式完成，
• 如果使用CompletionException例外完成，方法get()和get(long, TimeUnit)將拋出與對應的CompletionException中保持的相同原因的ExecutionException， 為了簡化大多數情況下的使用，此類還定義了方法join()和getNow(T) ，而是在這些情況下直接拋出CompletionException，

public <U> CompletionStage<U> thenApply(Function<? super T,? extends U> fn);
public <U> CompletionStage<U> thenApplyAsync(Function<? super T,? extends U> fn);
public <U> CompletionStage<U> thenApplyAsync(Function<? super T,? extends U> fn,Executor executor);

2、Guava 的 ListenableFuture

ListenableFuture 是對原有 Future 的增強，可以用于監聽 Future 任務的執行狀況，是執行成功還是執行失敗，并提供回應的介面用于對不同結果的處理，簡單來說就是他提供了一個回呼介面，在任務結束時可以主動去做通知呼叫，有點主動推的感覺，

public class ListenFutureTest {  

    public static void main(String[] args) {  
        testListenFuture();  
    }  

    public static void testListenFuture() {  
        System.out.println("主任務執行完，開始異步執行副任務1.....");  
        ListeningExecutorService pool = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(5));  
        ListenableFuture<String> future = pool.submit(new Task());  
        Futures.addCallback(future, new FutureCallback<String>() {  
            @Override  
            public void onSuccess(String result) {  
                System.out.println("成功,結果是:" + result);  
            }  

            @Override  
            public void onFailure(Throwable t) {  
                System.out.println("出錯,業務回滾或補償");  
            }  
        });  
        System.out.println("副本任務啟動,回歸主任務線，主業務正常回傳2.....");  
    }  
}  

class Task implements Callable<String> {  

    @Override  
    public String call() throws Exception {  
        TimeUnit.SECONDS.sleep(1);  
       // int a =1/0;  
        return "task done";  
    }  
} 

3、RxJava2

推薦使用RxJava2，具有背壓和默認執行緒調度功能，

    ExecutorService executor = Executors.newFixedThreadPool(2);
    System.out.println("MAIN: " + Thread.currentThread().getId());
    Callable<String> callable = () -> {
        System.out.println("callable [" + Thread.currentThread().getId() + "]: ");
        Path filePath = Paths.get("build.gradle");
        return Files.readAllLines(filePath).stream().flatMap(s -> Arrays.stream(s.split
                (""))).count() + "";
    };

    Future<String> future = executor.submit(callable);

    Consumer<String> onNext = v -> System.out
            .println("consumer[" + Thread.currentThread().getId() + "]:" + v);

    Flowable.fromCallable(callable).subscribe(onNext);
    Flowable.fromFuture(future).subscribe(onNext);
    System.out.println("END");

參考：

[RxJava2 回應式編程介紹] https://zouzhberk.github.io/rxjava-study/

掃描二維碼，關注公眾號“猿必過”

回復 “面試題” 自行領取吧，

微信群交流討論，請添加微信號：zyhui98，備注：面試題加群

本文由猿必過 YBG 發布

禁止未經授權轉載，違者依法追究相關法律責任

如需授權可聯系：[email protected]

標籤：Java

上一篇：Java 佇列同步器 AQS

下一篇：【福爾摩斯探案集】資料庫和應用程式時間出現14小時時差之迷