Java8 自定義CompletableFuture原理
Future 接口 的局限性有很多，其中一個就是需要主動的去詢問是否完成，如果等子線程的任務(wù)完成以后，通知我，那豈不是更好？
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
|
public class FutureInAction3 { public static void main(String[] args) { Future<String> future = invoke(() -> { try { Thread.sleep(10000L); return "I am Finished."; } catch (InterruptedException e) { return "I am Error"; } }); future.setCompletable(new Completable<String>() { @Override public void complete(String s) { System.out.println("complete called ---- " + s); } @Override public void exception(Throwable cause) { System.out.println("error"); cause.printStackTrace(); } }); System.out.println("....do something else ....."); System.out.println("try to get result ->" + future.get()); } private static <T> Future<T> invoke(Callable<T> callable) { AtomicReference<T> result = new AtomicReference<>(); AtomicBoolean finished = new AtomicBoolean(false); Future<T> future = new Future<T>() { private Completable<T> completable; @Override public T get() { return result.get(); } @Override public boolean isDone() { return finished.get(); } // 設(shè)置完成 @Override public void setCompletable(Completable<T> completable) { this.completable = completable; } // 獲取 @Override public Completable<T> getCompletable() { return completable; } }; Thread t = new Thread(() -> { try { T value = callable.action(); result.set(value); finished.set(true); if (future.getCompletable() != null) future.getCompletable().complete(value); } catch (Throwable cause) { if (future.getCompletable() != null) future.getCompletable().exception(cause); } }); t.start(); return future; } private interface Future<T> { T get(); boolean isDone(); // 1 void setCompletable(Completable<T> completable); // 2 Completable<T> getCompletable(); } private interface Callable<T> { T action(); } // 回調(diào)接口 private interface Completable<T> { void complete(T t); void exception(Throwable cause); }} |
CompleteFuture簡單使用
Java8 中的 completeFuture 是對 Future 的擴展實現(xiàn), 主要是為了彌補 Future 沒有相應(yīng)的回調(diào)機制的缺陷.
我們先看看 Java8 之前的 Future 的使用
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
|
package demos;import java.util.concurrent.ExecutionException;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Future;/** * @author djh on 2019/4/22 10:23 * @E-Mail [email protected] */public class Demo { public static void main(String[] args) throws ExecutionException, InterruptedException { ExecutorService cachePool = Executors.newCachedThreadPool(); Future<String> future = cachePool.submit(() -> { Thread.sleep(3000); return "異步任務(wù)計算結(jié)果!"; }); // 提交完異步任務(wù)后, 主線程可以繼續(xù)干一些其他的事情. doSomeThingElse(); // 為了獲取異步計算結(jié)果, 我們可以通過 future.get 和 輪詢機制來獲取. String result; // Get 方式會導(dǎo)致當(dāng)前線程阻塞, 這顯然違背了異步計算的初衷. // result = future.get(); // 輪詢方式雖然不會導(dǎo)致當(dāng)前線程阻塞, 但是會導(dǎo)致高額的 CPU 負載. long start = System.currentTimeMillis(); while (true) { if (future.isDone()) { break; } } System.out.println("輪詢耗時:" + (System.currentTimeMillis() - start)); result = future.get(); System.out.println("獲取到異步計算結(jié)果啦: " + result); cachePool.shutdown(); } private static void doSomeThingElse() { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("我的最重要的事情干完了, 我要獲取異步計算結(jié)果來執(zhí)行剩下的事情."); }} |
輸出：
我的最重要的事情干完了, 我要獲取異步計算結(jié)果來執(zhí)行剩下的事情.
輪詢耗時:2000
獲取到異步計算結(jié)果啦: 異步任務(wù)計算結(jié)果!Process finished with exit code 0
從上面的 Demo 中我們可以看出, future 在執(zhí)行異步任務(wù)時, 對于結(jié)果的獲取顯的不那么優(yōu)雅, 很多第三方庫就針對 Future 提供了回調(diào)式的接口以用來獲取異步計算結(jié)果, 如Google的: ListenableFuture, 而 Java8 所提供的 CompleteFuture 便是官方為了彌補這方面的不足而提供的 API.
下面簡單介紹用法
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
|
package demos;import java.util.concurrent.CompletableFuture;import java.util.concurrent.ExecutionException;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;/** * @author djh on 2019/5/1 20:26 * @E-Mail [email protected] */public class CompleteFutureDemo { public static void main(String[] args) throws ExecutionException, InterruptedException { CompletableFuture<String> completableFutureOne = new CompletableFuture<>(); ExecutorService cachePool = Executors.newCachedThreadPool(); cachePool.execute(() -> { try { Thread.sleep(3000); completableFutureOne.complete("異步任務(wù)執(zhí)行結(jié)果"); System.out.println(Thread.currentThread().getName()); } catch (InterruptedException e) { e.printStackTrace(); } }); // WhenComplete 方法返回的 CompletableFuture 仍然是原來的 CompletableFuture 計算結(jié)果. CompletableFuture<String> completableFutureTwo = completableFutureOne.whenComplete((s, throwable) -> { System.out.println("當(dāng)異步任務(wù)執(zhí)行完畢時打印異步任務(wù)的執(zhí)行結(jié)果: " + s); }); // ThenApply 方法返回的是一個新的 completeFuture. CompletableFuture<Integer> completableFutureThree = completableFutureTwo.thenApply(s -> { System.out.println("當(dāng)異步任務(wù)執(zhí)行結(jié)束時, 根據(jù)上一次的異步任務(wù)結(jié)果, 繼續(xù)開始一個新的異步任務(wù)!"); try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } return s.length(); }); System.out.println("阻塞方式獲取執(zhí)行結(jié)果:" + completableFutureThree.get()); cachePool.shutdown(); }} |
從上面的 Demo 中我們主要需要注意 thenApply 和 whenComplete 這兩個方法， 這兩個方法便是 CompleteFuture 中最具有意義的方法， 他們都會在 completeFuture 調(diào)用 complete 方法傳入異步計算結(jié)果時回調(diào)， 從而獲取到異步任務(wù)的結(jié)果.
相比之下 future 的阻塞和輪詢方式獲取異步任務(wù)的計算結(jié)果， CompleteFuture 獲取結(jié)果的方式就顯的優(yōu)雅的多。
以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持服務(wù)器之家。
原文鏈接：https://artisan.blog.csdn.net/article/details/115450097
