程序員注意了鼻由，帶你鳥瞰 Java 中4款常用的并發(fā)框架
為什么要寫這篇文章
幾年前 NoSQL 開始流行的時候，像其他團隊一樣厚棵，我們的團隊也熱衷于令人興奮的新東西蕉世，并且計劃替換一個應用程序的數(shù)據(jù)庫。 但是婆硬，當深入實現(xiàn)細節(jié)時狠轻，我們想起了一位智者曾經(jīng)說過的話：“細節(jié)決定成敗”。最終我們意識到 NoSQL 不是解決所有問題的銀彈彬犯，而 NoSQL vs RDMS 的答案是：“視情況而定”向楼。
類似地查吊，去年RxJava 和 Spring Reactor 這樣的并發(fā)庫加入了讓人充滿激情的語句，如異步非阻塞方法等湖蜕。為了避免再犯同樣的錯誤逻卖，我們嘗試評估諸如 ExecutorService、 RxJava昭抒、Disruptor 和 Akka 這些并發(fā)框架彼此之間的差異评也，以及如何確定各自框架的正確用法。
本文中用到的術(shù)語在這里有更詳細的描述灭返。
分析并發(fā)框架的示例用例
程序員注意了盗迟，帶你鳥瞰 Java 中4款常用的并發(fā)框架
快速更新線程配置
在開始比較并發(fā)框架的之前，讓我們快速復習一下如何配置最佳線程數(shù)以提高并行任務的性能婆殿。 這個理論適用于所有框架诈乒，并且在所有框架中使用相同的線程配置來度量性能。
- 對于內(nèi)存任務婆芦，線程的數(shù)量大約等于具有最佳性能的內(nèi)核的數(shù)量怕磨，盡管它可以根據(jù)各自處理器中的超線程特性進行一些更改。
- 例如消约，在8核機器中肠鲫，如果對應用程序的每個請求都必須在內(nèi)存中并行執(zhí)行4個任務，那么這臺機器上的負載應該保持為 @2 req/sec或粮，在 ThreadPool 中保持8個線程导饲。
2.對于 I/O 任務，ExecutorService 中配置的線程數(shù)應該取決于外部服務的延遲氯材。
- 與內(nèi)存中的任務不同渣锦，I/O 任務中涉及的線程將被阻塞，并處于等待狀態(tài)氢哮，直到外部服務響應或超時袋毙。 因此，當涉及 I/O 任務線程被阻塞時冗尤，應該增加線程的數(shù)量听盖，以處理來自并發(fā)請求的額外負載。
- I/O 任務的線程數(shù)應該以保守的方式增加裂七，因為處于活動狀態(tài)的許多線程帶來了上下文切換的成本皆看，這將影響應用程序的性能。 為了避免這種情況背零，應該根據(jù) I/O 任務中涉及的線程的等待時間按比例增加此機器的線程的確切數(shù)量以及負載腰吟。
性能測試結(jié)果
性能測試配置 GCP -> 處理器：Intel(R) Xeon(R) CPU @ 2.30GHz；架構(gòu)：x86_64徙瓶；CPU 內(nèi)核：8個（注意： 這些結(jié)果僅對該配置有意義毛雇，并不表示一個框架比另一個框架更好)录语。
程序員注意了，帶你鳥瞰 Java 中4款常用的并發(fā)框架
使用執(zhí)行器服務并行化 IO 任務
1. 何時使用禾乘？
如果一個應用程序部署在多個節(jié)點上，并且每個節(jié)點的 req/sec 小于可用的核心數(shù)量虽缕，那么 ExecutorService 可用于并行化任務始藕，更快地執(zhí)行代碼。
2. 什么時候適用氮趋？
如果一個應用程序部署在多個節(jié)點上伍派，并且每個節(jié)點的 req/sec 遠遠高于可用的核心數(shù)量，那么使用 ExecutorService 進一步并行化只會使情況變得更糟剩胁。
當外部服務延遲增加到 400ms 時诉植，性能測試結(jié)果如下（請求速率 @50 req/sec，8核）昵观。
程序員注意了晾腔，帶你鳥瞰 Java 中4款常用的并發(fā)框架
3. 所有任務按順序執(zhí)行示例
>// I/O 任務：調(diào)用外部服務
String posts = JsonService.getPosts();
String comments = JsonService.getComments();
String albums = JsonService.getAlbums();
String photos = JsonService.getPhotos();
// 合并來自外部服務的響應
// (內(nèi)存中的任務將作為此操作的一部分執(zhí)行)
int userId = new Random().nextInt(10) + 1;
String postsAndCommentsOfRandomUser = ResponseUtil.getPostsAndCommentsOfRandomUser(userId, posts, comments);
String albumsAndPhotosOfRandomUser = ResponseUtil.getAlbumsAndPhotosOfRandomUser(userId, albums, photos);
// 構(gòu)建最終響應并將其發(fā)送回客戶端
String response = postsAndCommentsOfRandomUser + albumsAndPhotosOfRandomUser;
return response;
</pre>
4. I/O 任務與 ExecutorService 并行執(zhí)行代碼示例
>// 添加 I/O 任務
List<Callable<String>> ioCallableTasks = new ArrayList<>();
ioCallableTasks.add(JsonService::getPosts);
ioCallableTasks.add(JsonService::getComments);
ioCallableTasks.add(JsonService::getAlbums);
ioCallableTasks.add(JsonService::getPhotos);
// 調(diào)用所有并行任務
ExecutorService ioExecutorService = CustomThreads.getExecutorService(ioPoolSize);
List<Future<String>> futuresOfIOTasks = ioExecutorService.invokeAll(ioCallableTasks);
// 獲取 I/O 操作(阻塞調(diào)用)結(jié)果
String posts = futuresOfIOTasks.get(0).get();
String comments = futuresOfIOTasks.get(1).get();
String albums = futuresOfIOTasks.get(2).get();
String photos = futuresOfIOTasks.get(3).get();
// 合并響應(內(nèi)存中的任務是此操作的一部分)
String postsAndCommentsOfRandomUser = ResponseUtil.getPostsAndCommentsOfRandomUser(userId, posts, comments);
String albumsAndPhotosOfRandomUser = ResponseUtil.getAlbumsAndPhotosOfRandomUser(userId, albums, photos);
// 構(gòu)建最終響應并將其發(fā)送回客戶端
return postsAndCommentsOfRandomUser + albumsAndPhotosOfRandomUser;
</pre>
使用執(zhí)行器服務并行化 IO 任務(CompletableFuture)
與上述情況類似：處理傳入請求的 HTTP 線程被阻塞，而 CompletableFuture 用于處理并行任務
何時使用啊犬？
如果沒有 AsyncResponse灼擂，性能與 ExecutorService 相同。 如果多個 API 調(diào)用必須異步并且鏈接起來觉至，那么這種方法更好(類似 Node 中的 Promises)剔应。
>ExecutorService ioExecutorService = CustomThreads.getExecutorService(ioPoolSize);
// I/O 任務
CompletableFuture<String> postsFuture = CompletableFuture.supplyAsync(JsonService::getPosts, ioExecutorService);
CompletableFuture<String> commentsFuture = CompletableFuture.supplyAsync(JsonService::getComments,
ioExecutorService);
CompletableFuture<String> albumsFuture = CompletableFuture.supplyAsync(JsonService::getAlbums,
ioExecutorService);
CompletableFuture<String> photosFuture = CompletableFuture.supplyAsync(JsonService::getPhotos,
ioExecutorService);
CompletableFuture.allOf(postsFuture, commentsFuture, albumsFuture, photosFuture).get();
// 從 I/O 任務(阻塞調(diào)用)獲得響應
String posts = postsFuture.get();
String comments = commentsFuture.get();
String albums = albumsFuture.get();
String photos = photosFuture.get();
// 合并響應(內(nèi)存中的任務將是此操作的一部分)
String postsAndCommentsOfRandomUser = ResponseUtil.getPostsAndCommentsOfRandomUser(userId, posts, comments);
String albumsAndPhotosOfRandomUser = ResponseUtil.getAlbumsAndPhotosOfRandomUser(userId, albums, photos);
// 構(gòu)建最終響應并將其發(fā)送回客戶端
return postsAndCommentsOfRandomUser + albumsAndPhotosOfRandomUser;
</pre>
使用 ExecutorService 并行處理所有任務
使用 ExecutorService 并行處理所有任務，并使用 @suspended AsyncResponse response 以非阻塞方式發(fā)送響應语御。
程序員注意了峻贮，帶你鳥瞰 Java 中4款常用的并發(fā)框架
- HTTP 線程處理傳入請求的連接，并將處理傳遞給 Executor Pool应闯，當所有任務完成后纤控，另一個 HTTP 線程將把響應發(fā)送回客戶端(異步非阻塞)。
- 性能下降原因:
- 在同步通信中孽锥，盡管 I/O 任務中涉及的線程被阻塞嚼黔，但是只要進程有額外的線程來承擔并發(fā)請求負載，它仍然處于運行狀態(tài)惜辑。
- 因此唬涧，以非阻塞方式保持線程所帶來的好處非常少，而且在此模式中處理請求所涉及的成本似乎很高盛撑。
- 通常碎节，對這里討論采用的例子使用異步非阻塞方法會降低應用程序的性能。
何時使用抵卫？
如果用例類似于服務器端聊天應用程序狮荔，在客戶端響應之前胎撇，線程不需要保持連接，那么異步殖氏、非阻塞方法比同步通信更受歡迎晚树。在這些用例中，系統(tǒng)資源可以通過異步雅采、非阻塞方法得到更好的利用爵憎，而不僅僅是等待。
>// 為異步執(zhí)行提交并行任務
ExecutorService ioExecutorService = CustomThreads.getExecutorService(ioPoolSize);
CompletableFuture<String> postsFuture = CompletableFuture.supplyAsync(JsonService::getPosts, ioExecutorService);
CompletableFuture<String> commentsFuture = CompletableFuture.supplyAsync(JsonService::getComments,
ioExecutorService);
CompletableFuture<String> albumsFuture = CompletableFuture.supplyAsync(JsonService::getAlbums,
ioExecutorService);
CompletableFuture<String> photosFuture = CompletableFuture.supplyAsync(JsonService::getPhotos,
ioExecutorService);
// 當 /posts API 返回響應時婚瓜，它將與來自 /comments API 的響應結(jié)合在一起
// 作為這個操作的一部分宝鼓，將執(zhí)行內(nèi)存中的一些任務
CompletableFuture<String> postsAndCommentsFuture = postsFuture.thenCombineAsync(commentsFuture,
(posts, comments) -> ResponseUtil.getPostsAndCommentsOfRandomUser(userId, posts, comments),
ioExecutorService);
// 當 /albums API 返回響應時，它將與來自 /photos API 的響應結(jié)合在一起
// 作為這個操作的一部分巴刻，將執(zhí)行內(nèi)存中的一些任務
CompletableFuture<String> albumsAndPhotosFuture = albumsFuture.thenCombineAsync(photosFuture,
(albums, photos) -> ResponseUtil.getAlbumsAndPhotosOfRandomUser(userId, albums, photos),
ioExecutorService);
// 構(gòu)建最終響應并恢復 http 連接愚铡，把響應發(fā)送回客戶端
postsAndCommentsFuture.thenAcceptBothAsync(albumsAndPhotosFuture, (s1, s2) -> {
LOG.info("Building Async Response in Thread " + Thread.currentThread().getName());
String response = s1 + s2;
asyncHttpResponse.resume(response);
}, ioExecutorService);
</pre>
RxJava
- 這與上面的情況類似，唯一的區(qū)別是 RxJava 提供了更好的 DSL 可以進行流式編程胡陪，下面的例子中沒有體現(xiàn)這一點沥寥。
- 性能優(yōu)于 CompletableFuture 處理并行任務。
何時使用柠座？
如果編碼的場景適合異步非阻塞方式营曼，那么可以首選 RxJava 或任何響應式開發(fā)庫。 還具有諸如 back-pressure 之類的附加功能愚隧，可以在生產(chǎn)者和消費者之間平衡負載蒂阱。
>int userId = new Random().nextInt(10) + 1;
ExecutorService executor = CustomThreads.getExecutorService(8);
// I/O 任務
Observable<String> postsObservable = Observable.just(userId).map(o -> JsonService.getPosts())
.subscribeOn(Schedulers.from(executor));
Observable<String> commentsObservable = Observable.just(userId).map(o -> JsonService.getComments())
.subscribeOn(Schedulers.from(executor));
Observable<String> albumsObservable = Observable.just(userId).map(o -> JsonService.getAlbums())
.subscribeOn(Schedulers.from(executor));
Observable<String> photosObservable = Observable.just(userId).map(o -> JsonService.getPhotos())
.subscribeOn(Schedulers.from(executor));
// 合并來自 /posts 和 /comments API 的響應
// 作為這個操作的一部分，將執(zhí)行內(nèi)存中的一些任務
Observable<String> postsAndCommentsObservable = Observable
.zip(postsObservable, commentsObservable,
(posts, comments) -> ResponseUtil.getPostsAndCommentsOfRandomUser(userId, posts, comments))
.subscribeOn(Schedulers.from(executor));
// 合并來自 /albums 和 /photos API 的響應
// 作為這個操作的一部分狂塘，將執(zhí)行內(nèi)存中的一些任務
Observable<String> albumsAndPhotosObservable = Observable
.zip(albumsObservable, photosObservable,
(albums, photos) -> ResponseUtil.getAlbumsAndPhotosOfRandomUser(userId, albums, photos))
.subscribeOn(Schedulers.from(executor));
// 構(gòu)建最終響應
Observable.zip(postsAndCommentsObservable, albumsAndPhotosObservable, (r1, r2) -> r1 + r2)
.subscribeOn(Schedulers.from(executor))
.subscribe((response) -> asyncResponse.resume(response), e -> asyncResponse.resume("error"));
</pre>
Disruptor
程序員注意了录煤，帶你鳥瞰 Java 中4款常用的并發(fā)框架
[Queue vs RingBuffer]
程序員注意了，帶你鳥瞰 Java 中4款常用的并發(fā)框架
- 在本例中荞胡，HTTP 線程將被阻塞妈踊，直到 disruptor 完成任務，并且使用 countdowlatch 將 HTTP 線程與 ExecutorService 中的線程同步泪漂。
- 這個框架的主要特點是在沒有任何鎖的情況下處理線程間通信廊营。在 ExecutorService 中，生產(chǎn)者和消費者之間的數(shù)據(jù)將通過 Queue傳遞萝勤，在生產(chǎn)者和消費者之間的數(shù)據(jù)傳輸過程中涉及到一個鎖露筒。 Disruptor 框架通過一個名為 Ring Buffer 的數(shù)據(jù)結(jié)構(gòu)（它是循環(huán)數(shù)組隊列的擴展版本）來處理這種生產(chǎn)者-消費者通信，并且不需要任何鎖敌卓。
- 這個庫不適用于我們在這里討論的這種用例慎式。僅出于好奇而添加。
何時使用？
Disruptor 框架在下列場合性能更好：與事件驅(qū)動的體系結(jié)構(gòu)一起使用瘪吏，或主要關(guān)注內(nèi)存任務的單個生產(chǎn)者和多個消費者癣防。
>static {
int userId = new Random().nextInt(10) + 1;
// 示例 Event-Handler; count down latch 用于使線程與 http 線程同步
EventHandler<Event> postsApiHandler = (event, sequence, endOfBatch) -> {
event.posts = JsonService.getPosts();
event.countDownLatch.countDown();
};
// 配置 Disputor 用于處理事件
DISRUPTOR.handleEventsWith(postsApiHandler, commentsApiHandler, albumsApiHandler)
.handleEventsWithWorkerPool(photosApiHandler1, photosApiHandler2)
.thenHandleEventsWithWorkerPool(postsAndCommentsResponseHandler1, postsAndCommentsResponseHandler2)
.handleEventsWithWorkerPool(albumsAndPhotosResponseHandler1, albumsAndPhotosResponseHandler2);
DISRUPTOR.start();
}
// 對于每個請求，在 RingBuffer 中發(fā)布一個事件:
Event event = null;
RingBuffer<Event> ringBuffer = DISRUPTOR.getRingBuffer();
long sequence = ringBuffer.next();
CountDownLatch countDownLatch = new CountDownLatch(6);
try {
event = ringBuffer.get(sequence);
event.countDownLatch = countDownLatch;
event.startTime = System.currentTimeMillis();
} finally {
ringBuffer.publish(sequence);
}
try {
event.countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
</pre>
Akka
程序員注意了掌眠，帶你鳥瞰 Java 中4款常用的并發(fā)框架
- Akka 庫的主要優(yōu)勢在于它擁有構(gòu)建分布式系統(tǒng)的本地支持蕾盯。
- 它運行在一個叫做 Actor System 的系統(tǒng)上。這個系統(tǒng)抽象了線程的概念蓝丙，Actor System 中的 Actor 通過異步消息進行通信刑枝，這類似于生產(chǎn)者和消費者之間的通信。
- 這種額外的抽象級別有助于 Actor System 提供諸如容錯迅腔、位置透明等特性。
- 使用正確的 Actor-to-Thread 策略靠娱，可以對該框架進行優(yōu)化沧烈，使其性能優(yōu)于上表所示的結(jié)果。 雖然它不能在單個節(jié)點上與傳統(tǒng)方法的性能匹敵像云，但是由于其構(gòu)建分布式和彈性系統(tǒng)的能力锌雀，仍然是首選。
示例代碼
>// 來自 controller :
Actors.masterActor.tell(new Master.Request("Get Response", event, Actors.workerActor), ActorRef.noSender());
// handler :
public Receive createReceive() {
return receiveBuilder().match(Request.class, request -> {
Event event = request.event; // Ideally, immutable data structures should be used here.
request.worker.tell(new JsonServiceWorker.Request("posts", event), getSelf());
request.worker.tell(new JsonServiceWorker.Request("comments", event), getSelf());
request.worker.tell(new JsonServiceWorker.Request("albums", event), getSelf());
request.worker.tell(new JsonServiceWorker.Request("photos", event), getSelf());
}).match(Event.class, e -> {
if (e.posts != null && e.comments != null & e.albums != null & e.photos != null) {
int userId = new Random().nextInt(10) + 1;
String postsAndCommentsOfRandomUser = ResponseUtil.getPostsAndCommentsOfRandomUser(userId, e.posts,
e.comments);
String albumsAndPhotosOfRandomUser = ResponseUtil.getAlbumsAndPhotosOfRandomUser(userId, e.albums,
e.photos);
String response = postsAndCommentsOfRandomUser + albumsAndPhotosOfRandomUser;
e.response = response;
e.countDownLatch.countDown();
}
}).build();
}
</pre>
總結(jié)
- 根據(jù)機器的負載決定 Executor 框架的配置迅诬，并檢查是否可以根據(jù)應用程序中并行任務的數(shù)量進行負載平衡腋逆。
- 對于大多數(shù)傳統(tǒng)應用程序來說，使用響應式開發(fā)庫或任何異步庫都會降低性能侈贷。只有當用例類似于服務器端聊天應用程序時惩歉，這個模式才有用，其中線程在客戶機響應之前不需要保留連接俏蛮。
- Disruptor 框架在與事件驅(qū)動的架構(gòu)模式一起使用時性能很好; 但是當 Disruptor 模式與傳統(tǒng)架構(gòu)混合使用時撑蚌，就我們在這里討論的用例而言，它并不符合標準搏屑。 這里需要注意的是争涌，Akka 和 Disruptor 庫值得單獨寫一篇文章，介紹如何使用它們來實現(xiàn)事件驅(qū)動的架構(gòu)模式辣恋。
- 這篇文章的源代碼可以在 GitHub 上找到亮垫。
最后
大家有任何問題都歡迎留言，之后也會給大家繼續(xù)帶來Java進階干貨伟骨！大家可以轉(zhuǎn)發(fā)關(guān)注一波~也可以點擊鏈接：https://shimo.im/docs/QzafqtfKU4MoODI9/read來領(lǐng)取我個人整理的一些比較有用的學習資料饮潦！
