1、什么是回調
設想一個情景，A是處理業務的一個步驟，A需要解決一個
問題，這時候A可以問B，讓B來告訴A答案，這期間，A可以繼續做自己的事情，而不用因為B做的事而阻塞。於是，我們想到給B設置一個線程，讓B去處理耗時的操作，然后處理完之后把結果告訴A。所以這個問題的要點就在於B處理完之后如何把結果告訴A。我們可以直接在A中寫一個方法對B處理完的結果進行處理，然后B處理完之后調用A這個方法。這樣A調用B去處理過程，B調用A的C方法去處理結果就叫做回調。
在正常的業務中使用同步線程，如果服務器每處理一個請求，就創建一個線程的話，會對服務器的資源造成浪費。因為這些線程可能會浪費時間在等待網絡傳輸，等待數據庫連接等其他事情上，真正處理業務邏輯的時間很短很短，但是其他線程在線程池滿了之后又會阻塞，等待前面的線程處理完成。而且，會出現一個奇怪的現象，客戶端的請求被阻塞，但是cpu的資源使用卻很低，大部分線程都浪費在處理其他事情上了。所以，這就導致服務器並發量不高。而異步，則可以解決這個問題。我們可以把需要用到cpu的業務處理使用異步來實現，這樣其他請求就不會被阻塞，而且cpu會保持比較高的使用率。綜上，可以使用回調來實現異步的方法。
2、Java回調范例
回調接口：
public interface CallBack {
/*
為什么要寫這個回調接口呢？
*因為可能不止主調A需要用到被調的處理過程，如果很多地方需要用到被調程序
* 那么傳入被調的方法就不可能只傳主調A類，所以要定義一個接口，
* 傳入被調的處理方法的參數就是這個接口對象
* */
public void solve(String result);
}
主調程序：
public class CallbackRequest implements Callback{
private CallbackResponse callbackResponse;
public CallbackRequest(CallbackResponse callbackResponse) {
this.callbackResponse = callbackResponse;
}
//主調需要解決一個問題，所以他把問題交給被調處理，被調單獨創建一個線程，不影響主調程序的運行
public void request(final String question){
System.out.println("主調程序問了一個問題");
new Thread(()->{
//B想要幫A處理東西，就必須知道誰讓自己處理的，所以要傳入a，也要知道a想處理什么，所以要傳入question
callbackResponse.handler(this, question);
}).start();
//A把要處理的事情交給b之后，就可以自己去玩耍了，或者去處理其他事情
afterAsk();
}
private void afterAsk(){
System.out.println("主調程序繼續處理其他事情");
}
@Override
public void solve(String result) {
System.out.println("被調程序接到答案后進行處理" + result);
}
}
被調程序：
public class CallbackResponse {
public void handler(Callback callback, String request) {
System.out.println(callback.getClass()+"問的問題是:"+ request);
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
String result="\n答案是2";
callback.solve(result);
}
}
測試：
public class CallbackTest {
public static void main(String[] args) {
CallbackResponse callbackResponse = new CallbackResponse();
CallbackRequest callbackRequest = new CallbackRequest(callbackResponse);
callbackRequest.request("1+1");
}
}
輸出：
主調程序問了一個問題
主調程序繼續處理其他事情
class javapratice.CallbackRequest問的問題是:1+1
被調程序接到答案后進行處理
答案是2
3、異步回調
異步回調的實現依賴於多線程或者多進程。軟件模塊之間總是存在着一定的接口，從調用方式上，可以把他們分為三類：同步調用、回調和異步調用。同步調用是一種阻塞式調用，調用方要等待對方執行完畢才返回，它是一種單向調用；回調是一種雙向調用模式，也就是說，被調用方在接口被調用時也會調用對方的接口；異步調用是一種類似消息或事件的機制，不過它的調用方向剛好相反，接口的服務在收到某種訊息或發生某種事件時，會主動通知客戶方（即調用客戶方的接口）。回調和異步調用的關系非常緊密，通常我們使用回調來實現異步消息的注冊，通過異步調用來實現消息的通知。
3.1、多線程中的“回調” （JDK8之前）
Java多線程中可以通過callable和future或futuretask結合來獲取線程執行后的返回值。實現方法是通過get方法來調用callable的call方法獲取返回值。其實這種方法本質上不是回調，回調要求的是任務完成以后被調用者主動回調調用者的接口，而這里是調用者主動使用get方法阻塞獲取返回值。一般情況下，我們會結合Callable和Future一起使用，通過ExecutorService的submit方法執行Callable，並返回Future。
//多線程中的“回調”
public class CallBackMultiThread {
//這里簡單地使用future和callable實現了線程執行完后
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executor = Executors.newCachedThreadPool();
Future<String> future = executor.submit(new Callable<String>() {
@Override
public String call() throws Exception {
System.out.println("call");
TimeUnit.SECONDS.sleep(1);
return "str";
}
});
//手動阻塞調用get通過call方法獲得返回值。
System.out.println(future.get());
//需要手動關閉，不然線程池的線程會繼續執行。
executor.shutdown();
//使用futuretask同時作為線程執行單元和數據請求單元。
FutureTask<Integer> futureTask = new FutureTask(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
System.out.println("dasds");
return new Random().nextInt();
}
});
new Thread(futureTask).start();
//阻塞獲取返回值
System.out.println(futureTask.get());
}
}
注：比起future.get()，其實更推薦使用get (long timeout, TimeUnit unit)方法，設置了超時時間可以防止程序無限制的等待future的結果。
很明顯，Future 接口提供了方法來檢測異步計算是否已經結束（使用isDone方法），等待異步操作結束，以及獲取計算的結果。但是這些特性還不足以編寫簡潔的並發代碼。比如，很難表述 Future 結果之間的依賴性；從文字描述上這很簡單，“當長時間計算任務完成時，請將該計算的結果通知到另一個長時間運行的計算任務，這兩個計算任務都完成后，將計算的結果與另一個查詢操作結果合並”。但是，使用Future中提供的方法完成這樣的操作又是另外一回事。
Future的主要缺點如下：
-
不支持手動完成：這個意思指的是，我提交了一個任務，但是執行太慢了，我通過其他路徑已經獲取到了任務結果，現在沒法把這個任務結果，通知到正在執行的線程，所以必須主動取消或者一直等待它執行完成。
-
不支持進一步的非阻塞調用：這個指的是我們通過Future的get方法會一直阻塞到任務完成，但是我還想在獲取任務之后，執行額外的任務，因為Future不支持回調函數，所以無法實現這個功能。
-
不支持鏈式調用：這個指的是對於Future的執行結果，我們想繼續傳到下一個Future處理使用，從而形成一個鏈式的pipline調用，這在Future中是沒法實現的。
-
不支持多個Future合並：比如我們有10個Future並行執行，我們想在所有的Future運行完畢之后，執行某些函數，是沒法通過Future實現的。
-
不支持異常處理：Future的API沒有任何的異常處理的api，所以在異步運行時，如果出了問題是不好定位的。
3.2、JDK8中新增的CompletableFuture
CompletableFuture在Java里面被用於異步編程，異步通常意味着非阻塞，可以使得我們的任務單獨運行在與主線程分離的其他線程中，並且通過回調可以在主線程中得到異步任務的執行狀態，是否完成，和是否異常等信息。CompletableFuture實現了Future, CompletionStage接口，實現了Future接口就可以兼容現在有線程池框架，而CompletionStage接口才是異步編程的接口抽象，里面定義多種異步方法，通過這兩者集合，從而打造出了強大的CompletableFuture類。
Future vs CompletableFuture
Futrue在Java里面，通常用來表示一個異步任務的引用，比如我們將任務提交到線程池里面，然后我們會得到一個Futrue，在Future里面有isDone方法來 判斷任務是否處理結束，還有get方法可以一直阻塞直到任務結束然后獲取結果，但整體來說這種方式，還是同步的，因為需要客戶端不斷阻塞等待或者不斷輪詢才能知道任務是否完成。
Java 8新增的CompletableFuture類正是吸收了所有Google Guava中ListenableFuture和SettableFuture的特征，還提供了其它強大的功能，讓Java擁有了完整的非阻塞編程模型：Future、Promise 和 Callback(在Java8之前，只有無Callback 的Future)。
CompletableFuture能夠將回調放到與任務不同的線程中執行，也能將回調作為繼續執行的同步函數，在與任務相同的線程中執行。它避免了傳統回調最大的問題，那就是能夠將控制流分離到不同的事件處理器中。
CompletableFuture彌補了Future模式的缺點。在異步的任務完成后，需要用其結果繼續操作時，無需等待。可以直接通過thenAccept、thenApply、thenCompose等方式將前面異步處理的結果交給另外一個異步事件處理線程來處理。
1)簡單使用CompletableFuture
1、一個最簡單的例子
在主線程里面創建一個CompletableFuture，然后主線程調用get方法會阻塞，最后我們在一個子線程中使其終止。
public class TestCompletableFuture {
public static void main(String[] args) throws Exception{
CompletableFuture<String> completableFuture=new CompletableFuture<String>();
Runnable runnable=new Runnable() {
@Override
public void run() {
try {
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName()+" 執行.....");
completableFuture.complete("success");//在子線程中完成主線程completableFuture的完成
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
Thread t1=new Thread(runnable);
t1.start();//啟動子線程
String result=completableFuture.get();//主線程阻塞，等待完成
System.out.println(Thread.currentThread().getName()+" 1x: "+result);
}
}
輸出結果：
Thread-0 執行.....
main 1x: success
2、運行一個簡單的沒有返回值的異步任務
public class TestCompletableFuture {
public static void main(String[] args) throws Exception{
CompletableFuture<Void> future=CompletableFuture.runAsync(new Runnable() {
@Override
public void run() {
try {
System.out.println(Thread.currentThread().getName()+"正在執行一個沒有返回值的異步任務。");
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
future.get();
System.out.println(Thread.currentThread().getName()+" 結束。");
}
}
輸出如下：
ForkJoinPool.commonPool-worker-1正在執行一個沒有返回值的異步任務。
main 結束。
從上面代碼我們可以看到CompletableFuture默認運行使用的是ForkJoin的的線程池。當然，你也可以用lambda表達式使得代碼更精簡。
3，運行一個有返回值的異步任務
public class TestCompletableFuture {
public static void main(String[] args) throws Exception{
CompletableFuture<String> future = CompletableFuture.supplyAsync(new Supplier<String>(){
@Override
public String get() {
try {
System.out.println(Thread.currentThread().getName()+"正在執行一個有返回值的異步任務。");
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "OK";
}
});
String result=future.get();
System.out.println(Thread.currentThread().getName()+" 結果："+result);
}
}
輸出結果：
ForkJoinPool.commonPool-worker-1正在執行一個有返回值的異步任務。
main 結果：OK
當然，上面默認的都是ForkJoinPool我們也可以換成Executor相關的Pool，其api都有支持如下：
static CompletableFuture<Void> runAsync(Runnable runnable) static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor) static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)
2)高級使用CompletableFuture
上面提到的幾種使用方法是使用異步編程最簡單的步驟，CompletableFuture.get()的方法會阻塞直到任務完成，這其實還是同步的概念，這對於一個異步系統是不夠的，因為真正的異步是需要支持回調函數，這樣以來，我們就可以直接在某個任務干完之后，接着執行回調里面的函數，從而做到真正的異步概念。在CompletableFuture里面，通過thenApply(), thenAccept()，thenRun()方法，來運行一個回調函數。
1、thenApply()
這個方法，其實用過函數式編程的人非常容易理解，類似於scala和spark的map算子，通過這個方法可以進行多次鏈式轉化並返回最終的加工結果。
看下面一個例子：
public class TestCompletableFuture {
public static void asyncCallback() throws ExecutionException, InterruptedException {
CompletableFuture<String> task=CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
System.out.println("線程" + Thread.currentThread().getName() + " supplyAsync");
return "123";
}
});
CompletableFuture<Integer> result1 = task.thenApply(number->{
System.out.println("線程" + Thread.currentThread().getName() + " thenApply1 ");
return Integer.parseInt(number);
});
CompletableFuture<Integer> result2 = result1.thenApply(number->{
System.out.println("線程" + Thread.currentThread().getName() + " thenApply2 ");
return number*2;
});
System.out.println("線程" + Thread.currentThread().getName()+" => "+result2.get());
}
public static void main(String[] args) throws Exception{
asyncCallback();
}
}
輸出結果：
線程ForkJoinPool.commonPool-worker-1 supplyAsync
線程main thenApply1
線程main thenApply2
線程main => 246
2、thenAccept()
這個方法，可以接受Futrue的一個返回值，但是本身不在返回任何值，適合用於多個callback函數的最后一步操作使用。例子如下：
public class TestCompletableFuture {
public static void asyncCallback() throws ExecutionException, InterruptedException {
CompletableFuture<String> task=CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
System.out.println(Thread.currentThread().getName()+" supplyAsync");
return "123";
}
});
CompletableFuture<Integer> chain1 = task.thenApply(number->{
System.out.println(Thread.currentThread().getName()+" thenApply1");
return Integer.parseInt(number);
});
CompletableFuture<Integer> chain2 = chain1.thenApply(number->{
System.out.println(Thread.currentThread().getName()+" thenApply2");
return number*2;
});
CompletableFuture<Void> result=chain2.thenAccept(product->{
System.out.println(Thread.currentThread().getName()+" thenAccept="+product);
});
result.get();
System.out.println(Thread.currentThread().getName()+" end");
}
public static void main(String[] args) throws Exception {
asyncCallback();
}
}
結果如下：
ForkJoinPool.commonPool-worker-1 supplyAsync
main thenApply1
main thenApply2
main thenAccept=246
main end
3、thenRun()
這個方法與上一個方法類似，一般也用於回調函數最后的執行，但這個方法不接受回調函數的返回值，純粹就代表執行任務的最后一個步驟：
public class TestCompletableFuture {
public static void asyncCallback() throws ExecutionException, InterruptedException {
CompletableFuture.supplyAsync(()->{
System.out.println(Thread.currentThread().getName()+"supplyAsync: 一階段任務");
return null;
}).thenRun(()->{
System.out.println(Thread.currentThread().getName()+"thenRun: 收尾任務");
}).get();
}
public static void main(String[] args) throws Exception {
asyncCallback();
}
}
結果：
ForkJoinPool.commonPool-worker-1supplyAsync: 一階段任務
mainthenRun: 收尾任務
這里注意，截止到目前，前面的例子代碼只會涉及兩個線程，一個是主線程一個是ForkJoinPool池的線程，但其實上面的每一步都是支持異步運行的，其api如下：
// thenApply() variants <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn) <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn) <U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)
我們看下改造后的一個例子：
public class TestCompletableFuture {
public static void asyncCallback() throws ExecutionException, InterruptedException {
CompletableFuture<String> ref1= CompletableFuture.supplyAsync(()->{
try {
System.out.println(Thread.currentThread().getName() + " supplyAsync開始執行任務1.... ");
TimeUnit.SECONDS.sleep(1);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + " supplyAsync: 任務1");
return null;
});
CompletableFuture<String> ref2= CompletableFuture.supplyAsync(()->{
try {
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + " thenApplyAsync: 任務2");
return null;
});
CompletableFuture<String> ref3=ref2.thenApplyAsync(value->{
System.out.println(Thread.currentThread().getName() +" thenApplyAsync: 任務2的子任務");
return " finish";
});
Thread.sleep(4000);
System.out.println(Thread.currentThread().getName() + ref3.get());
}
public static void main(String[] args) throws Exception {
asyncCallback();
}
}
輸出結果如下：
ForkJoinPool.commonPool-worker-1 supplyAsync開始執行任務1....
ForkJoinPool.commonPool-worker-2 thenApplyAsync: 任務2
ForkJoinPool.commonPool-worker-2 thenApplyAsync: 任務2的子任務
ForkJoinPool.commonPool-worker-1 supplyAsync: 任務1
main finish
我們可以看到，ForkJoin池的線程1，執行了前面的三個任務，但是第二個任務的子任務，因為我們了使用也異步提交所以它用的線程是ForkJoin池的線程2，最終由於main線程處執行了get是最后結束的。
還有一點需要注意：
ForkJoinPool所有的工作線程都是守護模式的，也就是說如果主線程退出，那么整個處理任務都會結束，而不管你當前的任務是否執行完。如果需要主線程等待結束，可采用ExecutorsThreadPool，如下：
ExecutorService pool = Executors.newFixedThreadPool(5);
final CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
... }, pool);
4、thenCompose()：合並兩個有依賴關系的CompletableFutures的執行結果
CompletableFutures在執行兩個依賴的任務合並時，會返回一個嵌套的結果列表，為了避免這種情況我們可以使用thenCompose來返回，直接獲取最頂層的結果數據即可：
public class TestCompletableFuture {
public static void asyncCompose() throws ExecutionException, InterruptedException {
CompletableFuture<String> future1=CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
return "1";
}
});
CompletableFuture<String>nestedResult = future1.thenCompose(value->
CompletableFuture.supplyAsync(()->{
return value+"2";
}));
System.out.println(nestedResult.get());
}
public static void main(String[] args) throws Exception {
asyncCompose();
}
}
輸出結果:12
5、thenCombine：組合兩個沒有依賴關系的CompletableFutures任務
public class TestCompletableFuture {
public static void asyncCombine() throws ExecutionException, InterruptedException {
CompletableFuture<Double> d1= CompletableFuture.supplyAsync(new Supplier<Double>() {
@Override
public Double get() {
return 1d;
}
});
CompletableFuture<Double> d2= CompletableFuture.supplyAsync(new Supplier<Double>() {
@Override
public Double get() {
return 2d;
}
});
CompletableFuture<Double> result= d1.thenCombine(d2,(number1,number2)->{
return number1+number2;
});
System.out.println(result.get());
}
public static void main(String[] args) throws Exception {
asyncCombine();
}
}
輸出結果：3d
6、合並多個任務的結果allOf與anyOf
上面說的是兩個任務的合並，那么多個任務需要使用allOf或者anyOf方法。allOf適用於，你有一系列獨立的future任務，你想等其所有的任務執行完后做一些事情。舉個例子，比如我想下載100個網頁，傳統的串行，性能肯定不行，這里我們采用異步模式，同時對100個網頁進行下載，當所有的任務下載完成之后，我們想判斷每個網頁是否包含某個關鍵詞。
下面我們通過隨機數來模擬上面的這個場景如下：
public class TestCompletableFuture {
public static void mutilTaskTest() throws ExecutionException, InterruptedException {
//添加n個任務
CompletableFuture<Double> array[]=new CompletableFuture[3];
for ( int i = 0; i < 3; i++) {
array[i]=CompletableFuture.supplyAsync(new Supplier<Double>() {
@Override
public Double get() {
return Math.random();
}
});
}
//獲取結果的方式一
// CompletableFuture.allOf(array).get();
// for(CompletableFuture<Double> cf:array){
// if(cf.get()>0.6){
// System.out.println(cf.get());
// }
// }
//獲取結果的方式二，過濾大於指定數字，在收集輸出
List<Double> rs= Stream.of(array).map(CompletableFuture::join).filter(number->number>0.6).collect(Collectors.toList());
System.out.println(rs);
}
public static void main(String[] args) throws Exception {
mutilTaskTest();
}
}
結果如下（結果可能不一致）：
[0.85538057702618, 0.7692532053269862, 0.6441387373310598]
注意其中的join方法和get方法類似，僅僅在於在Future不能正常完成的時候拋出一個unchecked的exception，這可以確保它用在Stream的map方法中，直接使用get是沒法在map里面運行的。
anyOf方法，也比較簡單，意思就是只要在多個future里面有一個返回，整個任務就可以結束，而不需要等到每一個future結束。
public class TestCompletableFuture {
public static void mutilTaskTest() throws ExecutionException, InterruptedException {
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
try {
TimeUnit.SECONDS.sleep(4);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "wait 4 seconds";
}
});
CompletableFuture<String> f2 = CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "wait 2 seconds";
}
});
CompletableFuture<String> f3 = CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
try {
TimeUnit.SECONDS.sleep(4);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "wait 10 seconds";
}
});
CompletableFuture<Object> result = CompletableFuture.anyOf(f1, f2, f3);
System.out.println(result.get());
}
public static void main(String[] args) throws Exception {
mutilTaskTest();
}
}
輸出結果：
wait 2 seconds
注意由於Anyof返回的是其中任意一個Future所以這里沒有明確的返回類型，統一使用Object接受，留給使用端處理。
7、exceptionally異常處理
異常處理是異步計算的一個重要環節，下面看看如何在CompletableFuture中使用：
public class TestCompletableFuture {
public static void exceptionProcess() throws ExecutionException, InterruptedException {
int age=-1;
CompletableFuture<String> task= CompletableFuture.supplyAsync(new Supplier<String>(){
@Override
public String get(){
if(age<0){
throw new IllegalArgumentException("性別必須大於0");
}
if(age<18){
return "未成年人";
}
return "成年人";
}
}).exceptionally(ex->{
System.out.println(ex.getMessage());
return "發生 異常"+ex.getMessage();
});
System.out.println(task.get());
}
public static void main(String[] args) throws Exception {
exceptionProcess();
}
}
結果如下：
java.lang.IllegalArgumentException: 性別必須大於0
發生 異常java.lang.IllegalArgumentException: 性別必須大於0
此外還有另外一種異常捕捉方法handle，無論發生異常都會執行，示例如下：
public class TestCompletableFuture {
public static void exceptionProcess() throws ExecutionException, InterruptedException {
int age = -10;
CompletableFuture<String> task= CompletableFuture.supplyAsync(new Supplier<String>(){
@Override
public String get(){
if(age<0){
throw new IllegalArgumentException("性別必須大於0");
}
if(age<18){
return "未成年人";
}
return "成年人";
}
}).handle((res,ex)->{
System.out.println("執行handle");
if(ex!=null){
System.out.println("發生異常");
return "發生 異常"+ex.getMessage();
}
return res;
});
System.out.println(task.get());
}
public static void main(String[] args) throws Exception {
exceptionProcess();
}
}
輸出結果：
執行handle
發生異常
發生 異常java.lang.IllegalArgumentException: 性別必須大於0
注意上面的方法如果正常執行，也會執行handle方法。
3.3、JDK9 CompletableFuture 類增強的主要內容
（1）支持對異步方法的超時調用
-
orTimeout()
-
completeOnTimeout()
（2）支持延遲調用
-
Executor delayedExecutor(long delay, TimeUnit unit, Executor executor)
-
Executor delayedExecutor(long delay, TimeUnit unit)