日本综合一区二区|亚洲中文天堂综合|日韩欧美自拍一区|男女精品天堂一区|欧美自拍第6页亚洲成人精品一区|亚洲黄色天堂一区二区成人|超碰91偷拍第一页|日韩av夜夜嗨中文字幕|久久蜜综合视频官网|精美人妻一区二区三区

RELATEED CONSULTING
相關(guān)咨詢
選擇下列產(chǎn)品馬上在線溝通
服務(wù)時(shí)間:8:30-17:00
你可能遇到了下面的問題
關(guān)閉右側(cè)工具欄

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營(yíng)銷解決方案
鳥瞰Java并發(fā)框架

1. 為什么要寫這篇文章

創(chuàng)新互聯(lián)是一家專注于成都網(wǎng)站設(shè)計(jì)、成都網(wǎng)站建設(shè)與策劃設(shè)計(jì),青原網(wǎng)站建設(shè)哪家好?創(chuàng)新互聯(lián)做網(wǎng)站,專注于網(wǎng)站建設(shè)10年,網(wǎng)設(shè)計(jì)領(lǐng)域的專業(yè)建站公司;建站業(yè)務(wù)涵蓋:青原等地區(qū)。青原做網(wǎng)站價(jià)格咨詢:18982081108

幾年前 NoSQL 開始流行的時(shí)候,像其他團(tuán)隊(duì)一樣,我們的團(tuán)隊(duì)也熱衷于令人興奮的新東西,并且計(jì)劃替換一個(gè)應(yīng)用程序的數(shù)據(jù)庫(kù)。但是,當(dāng)深入實(shí)現(xiàn)細(xì)節(jié)時(shí),我們想起了一位智者曾經(jīng)說過的話:“細(xì)節(jié)決定成敗”。最終我們意識(shí)到 NoSQL 不是解決所有問題的銀彈,而 NoSQL vs RDMS 的答案是:“視情況而定”。類似地,去年RxJava 和 Spring Reactor 這樣的并發(fā)庫(kù)加入了讓人充滿激情的語(yǔ)句,如異步非阻塞方法等。為了避免再犯同樣的錯(cuò)誤,我們嘗試評(píng)估諸如 ExecutorService、 RxJava、Disruptor 和 Akka 這些并發(fā)框架彼此之間的差異,以及如何確定各自框架的正確用法。

本文中用到的術(shù)語(yǔ)在這里有更詳細(xì)的描述。

2. 分析并發(fā)框架的示例用例

3. 快速更新線程配置

在開始比較并發(fā)框架的之前,讓我們快速?gòu)?fù)習(xí)一下如何配置最優(yōu)線程數(shù)以提高并行任務(wù)的性能。這個(gè)理論適用于所有框架,并且在所有框架中使用相同的線程配置來度量性能。

  • 對(duì)于內(nèi)存任務(wù),線程的數(shù)量大約等于具有***性能的內(nèi)核的數(shù)量,盡管它可以根據(jù)各自處理器中的超線程特性進(jìn)行一些更改。
    • 例如,在8核機(jī)器中,如果對(duì)應(yīng)用程序的每個(gè)請(qǐng)求都必須在內(nèi)存中并行執(zhí)行4個(gè)任務(wù),那么這臺(tái)機(jī)器上的負(fù)載應(yīng)該保持為 @2 req/sec,在 ThreadPool 中保持8個(gè)線程。
  • 對(duì)于 I/O 任務(wù),ExecutorService 中配置的線程數(shù)應(yīng)該取決于外部服務(wù)的延遲。
    • 與內(nèi)存中的任務(wù)不同,I/O 任務(wù)中涉及的線程將被阻塞,并處于等待狀態(tài),直到外部服務(wù)響應(yīng)或超時(shí)。因此,當(dāng)涉及 I/O 任務(wù)線程被阻塞時(shí),應(yīng)該增加線程的數(shù)量,以處理來自并發(fā)請(qǐng)求的額外負(fù)載。
    • I/O 任務(wù)的線程數(shù)應(yīng)該以保守的方式增加,因?yàn)樘幱诨顒?dòng)狀態(tài)的許多線程帶來了上下文切換的成本,這將影響應(yīng)用程序的性能。為了避免這種情況,應(yīng)該根據(jù) I/O 任務(wù)中涉及的線程的等待時(shí)間按比例增加此機(jī)器的線程的確切數(shù)量以及負(fù)載。

參考: http://baddotrobot.com/blog/2013/06/01/optimum-number-of-threads/

4. 性能測(cè)試結(jié)果

性能測(cè)試配置 GCP -> 處理器:Intel(R) Xeon(R) CPU @ 2.30GHz;架構(gòu):x86_64;CPU 內(nèi)核:8個(gè)(注意:這些結(jié)果僅對(duì)該配置有意義,并不表示一個(gè)框架比另一個(gè)框架更好)。

5. 使用執(zhí)行器服務(wù)并行化 IO 任務(wù)

5.1 何時(shí)使用?

如果一個(gè)應(yīng)用程序部署在多個(gè)節(jié)點(diǎn)上,并且每個(gè)節(jié)點(diǎn)的 req/sec 小于可用的核心數(shù)量,那么 ExecutorService 可用于并行化任務(wù),更快地執(zhí)行代碼。

5.2 什么時(shí)候適用?

如果一個(gè)應(yīng)用程序部署在多個(gè)節(jié)點(diǎn)上,并且每個(gè)節(jié)點(diǎn)的 req/sec 遠(yuǎn)遠(yuǎn)高于可用的核心數(shù)量,那么使用 ExecutorService 進(jìn)一步并行化只會(huì)使情況變得更糟。

當(dāng)外部服務(wù)延遲增加到 400ms 時(shí),性能測(cè)試結(jié)果如下(請(qǐng)求速率 @50 req/sec,8核)。

5.3 所有任務(wù)按順序執(zhí)行示例

 
 
 
 
  1. // I/O 任務(wù):調(diào)用外部服務(wù) 
  2. String posts = JsonService.getPosts(); 
  3. String comments = JsonService.getComments(); 
  4. String albums = JsonService.getAlbums(); 
  5. String photos = JsonService.getPhotos(); 
  6.  
  7. // 合并來自外部服務(wù)的響應(yīng) 
  8. // (內(nèi)存中的任務(wù)將作為此操作的一部分執(zhí)行) 
  9. int userId = new Random().nextInt(10) + 1; 
  10. String postsAndCommentsOfRandomUser = ResponseUtil.getPostsAndCommentsOfRandomUser(userId, posts, comments); 
  11. String albumsAndPhotosOfRandomUser = ResponseUtil.getAlbumsAndPhotosOfRandomUser(userId, albums, photos); 
  12.  
  13. // 構(gòu)建最終響應(yīng)并將其發(fā)送回客戶端 
  14. String response = postsAndCommentsOfRandomUser + albumsAndPhotosOfRandomUser; 
  15. return response; 

5.4 I/O 任務(wù)與 ExecutorService 并行執(zhí)行代碼示例

 
 
 
 
  1. // 添加 I/O 任務(wù) 
  2. List> ioCallableTasks = new ArrayList<>(); 
  3. ioCallableTasks.add(JsonService::getPosts); 
  4. ioCallableTasks.add(JsonService::getComments); 
  5. ioCallableTasks.add(JsonService::getAlbums); 
  6. ioCallableTasks.add(JsonService::getPhotos); 
  7.  
  8. // 調(diào)用所有并行任務(wù) 
  9. ExecutorService ioExecutorService = CustomThreads.getExecutorService(ioPoolSize); 
  10. List> futuresOfIOTasks = ioExecutorService.invokeAll(ioCallableTasks); 
  11.  
  12. // 獲取 I/O  操作(阻塞調(diào)用)結(jié)果 
  13. String posts = futuresOfIOTasks.get(0).get(); 
  14. String comments = futuresOfIOTasks.get(1).get(); 
  15. String albums = futuresOfIOTasks.get(2).get(); 
  16. String photos = futuresOfIOTasks.get(3).get(); 
  17.  
  18. // 合并響應(yīng)(內(nèi)存中的任務(wù)是此操作的一部分) 
  19. String postsAndCommentsOfRandomUser = ResponseUtil.getPostsAndCommentsOfRandomUser(userId, posts, comments); 
  20. String albumsAndPhotosOfRandomUser = ResponseUtil.getAlbumsAndPhotosOfRandomUser(userId, albums, photos); 
  21.  
  22. // 構(gòu)建最終響應(yīng)并將其發(fā)送回客戶端 
  23. return postsAndCommentsOfRandomUser + albumsAndPhotosOfRandomUser; 

6. 使用執(zhí)行器服務(wù)并行化 IO 任務(wù)(CompletableFuture)

與上述情況類似:處理傳入請(qǐng)求的 HTTP 線程被阻塞,而 CompletableFuture 用于處理并行任務(wù)。

6.1 何時(shí)使用?

如果沒有 AsyncResponse,性能與 ExecutorService 相同。如果多個(gè) API 調(diào)用必須異步并且鏈接起來,那么這種方法更好(類似 Node 中的 Promises)。

 
 
 
 
  1. ExecutorService ioExecutorService = CustomThreads.getExecutorService(ioPoolSize); 
  2.  
  3. // I/O 任務(wù) 
  4. CompletableFuture postsFuture = CompletableFuture.supplyAsync(JsonService::getPosts, ioExecutorService); 
  5. CompletableFuture commentsFuture = CompletableFuture.supplyAsync(JsonService::getComments, 
  6.     ioExecutorService); 
  7. CompletableFuture albumsFuture = CompletableFuture.supplyAsync(JsonService::getAlbums, 
  8.     ioExecutorService); 
  9. CompletableFuture photosFuture = CompletableFuture.supplyAsync(JsonService::getPhotos, 
  10.     ioExecutorService); 
  11. CompletableFuture.allOf(postsFuture, commentsFuture, albumsFuture, photosFuture).get(); 
  12.  
  13. // 從 I/O 任務(wù)(阻塞調(diào)用)獲得響應(yīng) 
  14. String posts = postsFuture.get(); 
  15. String comments = commentsFuture.get(); 
  16. String albums = albumsFuture.get(); 
  17. String photos = photosFuture.get(); 
  18.  
  19. // 合并響應(yīng)(內(nèi)存中的任務(wù)將是此操作的一部分) 
  20. String postsAndCommentsOfRandomUser = ResponseUtil.getPostsAndCommentsOfRandomUser(userId, posts, comments); 
  21. String albumsAndPhotosOfRandomUser = ResponseUtil.getAlbumsAndPhotosOfRandomUser(userId, albums, photos); 
  22.  
  23. // 構(gòu)建最終響應(yīng)并將其發(fā)送回客戶端 
  24. return postsAndCommentsOfRandomUser + albumsAndPhotosOfRandomUser; 

7. 使用 ExecutorService 并行處理所有任務(wù)

使用 ExecutorService 并行處理所有任務(wù),并使用 @suspended AsyncResponse response 以非阻塞方式發(fā)送響應(yīng)。

圖片來自 http://tutorials.jenkov.com/java-nio/nio-vs-io.html

  • HTTP 線程處理傳入請(qǐng)求的連接,并將處理傳遞給 Executor Pool,當(dāng)所有任務(wù)完成后,另一個(gè) HTTP 線程將把響應(yīng)發(fā)送回客戶端(異步非阻塞)。
  • 性能下降原因:
    • 在同步通信中,盡管 I/O 任務(wù)中涉及的線程被阻塞,但是只要進(jìn)程有額外的線程來承擔(dān)并發(fā)請(qǐng)求負(fù)載,它仍然處于運(yùn)行狀態(tài)。
    • 因此,以非阻塞方式保持線程所帶來的好處非常少,而且在此模式中處理請(qǐng)求所涉及的成本似乎很高。
    • 通常,對(duì)這里討論采用的例子使用異步非阻塞方法會(huì)降低應(yīng)用程序的性能。

7.1 何時(shí)使用?

如果用例類似于服務(wù)器端聊天應(yīng)用程序,在客戶端響應(yīng)之前,線程不需要保持連接,那么異步、非阻塞方法比同步通信更受歡迎。在這些用例中,系統(tǒng)資源可以通過異步、非阻塞方法得到更好的利用,而不僅僅是等待。

 
 
 
 
  1. // 為異步執(zhí)行提交并行任務(wù) 
  2. ExecutorService ioExecutorService = CustomThreads.getExecutorService(ioPoolSize); 
  3. CompletableFuture postsFuture = CompletableFuture.supplyAsync(JsonService::getPosts, ioExecutorService); 
  4. CompletableFuture commentsFuture = CompletableFuture.supplyAsync(JsonService::getComments, 
  5. ioExecutorService); 
  6. CompletableFuture albumsFuture = CompletableFuture.supplyAsync(JsonService::getAlbums, 
  7. ioExecutorService); 
  8. CompletableFuture photosFuture = CompletableFuture.supplyAsync(JsonService::getPhotos, 
  9. ioExecutorService); 
  10.  
  11. // 當(dāng) /posts API 返回響應(yīng)時(shí),它將與來自 /comments API 的響應(yīng)結(jié)合在一起 
  12. // 作為這個(gè)操作的一部分,將執(zhí)行內(nèi)存中的一些任務(wù) 
  13. CompletableFuture postsAndCommentsFuture = postsFuture.thenCombineAsync(commentsFuture, 
  14. (posts, comments) -> ResponseUtil.getPostsAndCommentsOfRandomUser(userId, posts, comments), 
  15. ioExecutorService); 
  16.  
  17. // 當(dāng) /albums API 返回響應(yīng)時(shí),它將與來自 /photos API 的響應(yīng)結(jié)合在一起 
  18. // 作為這個(gè)操作的一部分,將執(zhí)行內(nèi)存中的一些任務(wù) 
  19. CompletableFuture albumsAndPhotosFuture = albumsFuture.thenCombineAsync(photosFuture, 
  20. (albums, photos) -> ResponseUtil.getAlbumsAndPhotosOfRandomUser(userId, albums, photos), 
  21. ioExecutorService); 
  22.  
  23. // 構(gòu)建最終響應(yīng)并恢復(fù) http 連接,把響應(yīng)發(fā)送回客戶端 
  24. postsAndCommentsFuture.thenAcceptBothAsync(albumsAndPhotosFuture, (s1, s2) -> { 
  25. LOG.info("Building Async Response in Thread " + Thread.currentThread().getName()); 
  26. String response = s1 + s2; 
  27. asyncHttpResponse.resume(response); 
  28. }, ioExecutorService); 

8. RxJava

  • 這與上面的情況類似,唯一的區(qū)別是 RxJava 提供了更好的 DSL 可以進(jìn)行流式編程,下面的例子中沒有體現(xiàn)這一點(diǎn)。
  • 性能優(yōu)于 CompletableFuture 處理并行任務(wù)。

8.1 何時(shí)使用?

如果編碼的場(chǎng)景適合異步非阻塞方式,那么可以*** RxJava 或任何響應(yīng)式開發(fā)庫(kù)。還具有諸如 back-pressure 之類的附加功能,可以在生產(chǎn)者和消費(fèi)者之間平衡負(fù)載。

 
 
 
 
  1. int userId = new Random().nextInt(10) + 1; 
  2. ExecutorService executor = CustomThreads.getExecutorService(8); 
  3.  
  4. // I/O 任務(wù) 
  5. Observable postsObservable = Observable.just(userId).map(o -> JsonService.getPosts()) 
  6. .subscribeOn(Schedulers.from(executor)); 
  7. Observable commentsObservable = Observable.just(userId).map(o -> JsonService.getComments()) 
  8. .subscribeOn(Schedulers.from(executor)); 
  9. Observable albumsObservable = Observable.just(userId).map(o -> JsonService.getAlbums()) 
  10. .subscribeOn(Schedulers.from(executor)); 
  11. Observable photosObservable = Observable.just(userId).map(o -> JsonService.getPhotos()) 
  12. .subscribeOn(Schedulers.from(executor)); 
  13.  
  14. // 合并來自 /posts 和 /comments API 的響應(yīng) 
  15. // 作為這個(gè)操作的一部分,將執(zhí)行內(nèi)存中的一些任務(wù) 
  16. Observable postsAndCommentsObservable = Observable 
  17. .zip(postsObservable, commentsObservable, 
  18. (posts, comments) -> ResponseUtil.getPostsAndCommentsOfRandomUser(userId, posts, comments)) 
  19. .subscribeOn(Schedulers.from(executor)); 
  20.  
  21. // 合并來自 /albums 和 /photos API 的響應(yīng) 
  22. // 作為這個(gè)操作的一部分,將執(zhí)行內(nèi)存中的一些任務(wù) 
  23. Observable albumsAndPhotosObservable = Observable 
  24. .zip(albumsObservable, photosObservable, 
  25. (albums, photos) -> ResponseUtil.getAlbumsAndPhotosOfRandomUser(userId, albums, photos)) 
  26. .subscribeOn(Schedulers.from(executor)); 
  27.  
  28. // 構(gòu)建最終響應(yīng) 
  29. Observable.zip(postsAndCommentsObservable, albumsAndPhotosObservable, (r1, r2) -> r1 + r2) 
  30. .subscribeOn(Schedulers.from(executor)) 
  31. .subscribe((response) -> asyncResponse.resume(response), e -> asyncResponse.resume("error")); 

9. Disruptor

[Queue vs RingBuffer]

圖片1:http://tutorials.jenkov.com/java-concurrency/blocking-queues.html

圖片2:https://www.baeldung.com/lmax-disruptor-concurrency

  • 在本例中,HTTP 線程將被阻塞,直到 disruptor 完成任務(wù),并且使用 countdowlatch 將 HTTP 線程與 ExecutorService 中的線程同步。
  • 這個(gè)框架的主要特點(diǎn)是在沒有任何鎖的情況下處理線程間通信。在 ExecutorService 中,生產(chǎn)者和消費(fèi)者之間的數(shù)據(jù)將通過 Queue傳遞,在生產(chǎn)者和消費(fèi)者之間的數(shù)據(jù)傳輸過程中涉及到一個(gè)鎖。Disruptor 框架通過一個(gè)名為 Ring Buffer 的數(shù)據(jù)結(jié)構(gòu)(它是循環(huán)數(shù)組隊(duì)列的擴(kuò)展版本)來處理這種生產(chǎn)者-消費(fèi)者通信,并且不需要任何鎖。
  • 這個(gè)庫(kù)不適用于我們?cè)谶@里討論的這種用例。僅出于好奇而添加。

9.1 何時(shí)使用?

Disruptor 框架在下列場(chǎng)合性能更好:與事件驅(qū)動(dòng)的體系結(jié)構(gòu)一起使用,或主要關(guān)注內(nèi)存任務(wù)的單個(gè)生產(chǎn)者和多個(gè)消費(fèi)者。

 
 
 
 
  1. static { 
  2.     int userId = new Random().nextInt(10) + 1; 
  3.  
  4.     // 示例 Event-Handler; count down latch 用于使線程與 http 線程同步 
  5.     EventHandler postsApiHandler = (event, sequence, endOfBatch) -> { 
  6.         event.posts = JsonService.getPosts(); 
  7.         event.countDownLatch.countDown(); 
  8.     }; 
  9.  
  10.     // 配置 Disputor 用于處理事件 
  11.     DISRUPTOR.handleEventsWith(postsApiHandler, commentsApiHandler, albumsApiHandler) 
  12.     .handleEventsWithWorkerPool(photosApiHandler1, photosApiHandler2) 
  13.     .thenHandleEventsWithWorkerPool(postsAndCommentsResponseHandler1, postsAndCommentsResponseHandler2) 
  14.     .handleEventsWithWorkerPool(albumsAndPhotosResponseHandler1, albumsAndPhotosResponseHandler2); 
  15.     DISRUPTOR.start(); 
  16.  
  17. // 對(duì)于每個(gè)請(qǐng)求,在 RingBuffer 中發(fā)布一個(gè)事件: 
  18. Event event = null; 
  19. RingBuffer ringBuffer = DISRUPTOR.getRingBuffer(); 
  20. long sequence = ringBuffer.next(); 
  21. CountDownLatch countDownLatch = new CountDownLatch(6); 
  22. try { 
  23.     event = ringBuffer.get(sequence); 
  24.     event.countDownLatch = countDownLatch; 
  25.     event.startTime = System.currentTimeMillis(); 
  26. } finally { 
  27.     ringBuffer.publish(sequence); 
  28. try { 
  29.     event.countDownLatch.await(); 
  30. } catch (InterruptedException e) { 
  31.     e.printStackTrace(); 

10. Akka

圖片來自:https://blog.codecentric.de/en/2015/08/introduction-to-akka-actors/

  • Akka 庫(kù)的主要優(yōu)勢(shì)在于它擁有構(gòu)建分布式系統(tǒng)的本地支持。
  • 它運(yùn)行在一個(gè)叫做 Actor System 的系統(tǒng)上。這個(gè)系統(tǒng)抽象了線程的概念,Actor System 中的 Actor 通過異步消息進(jìn)行通信,這類似于生產(chǎn)者和消費(fèi)者之間的通信。
  • 這種額外的抽象級(jí)別有助于 Actor System 提供諸如容錯(cuò)、位置透明等特性。
  • 使用正確的 Actor-to-Thread 策略,可以對(duì)該框架進(jìn)行優(yōu)化,使其性能優(yōu)于上表所示的結(jié)果。雖然它不能在單個(gè)節(jié)點(diǎn)上與傳統(tǒng)方法的性能匹敵,但是由于其構(gòu)建分布式和彈性系統(tǒng)的能力,仍然是***。

10.1 示例代碼

 
 
 
 
  1. // 來自 controller : 
  2. Actors.masterActor.tell(new Master.Request("Get Response", event, Actors.workerActor), ActorRef.noSender()); 
  3.  
  4. // handler : 
  5. public Receive createReceive() { 
  6.     return receiveBuilder().match(Request.class, request -> { 
  7.     Event event = request.event; // Ideally, immutable data structures should be used here. 
  8.     request.worker.tell(new JsonServiceWorker.Request("posts", event), getSelf()); 
  9.     request.worker.tell(new JsonServiceWorker.Request("comments", event), getSelf()); 
  10.     request.worker.tell(new JsonServiceWorker.Request("albums", event), getSelf()); 
  11.     request.worker.tell(new JsonServiceWorker.Request("photos", event), getSelf()); 
  12.     }).match(Event.class, e -> { 
  13.     if (e.posts != null && e.comments != null & e.albums != null & e.photos != null) { 
  14.     int userId = new Random().nextInt(10) + 1; 
  15.     String postsAndCommentsOfRandomUser = ResponseUtil.getPostsAndCommentsOfRandomUser(userId, e.posts, 
  16.     e.comments); 
  17.     String albumsAndPhotosOfRandomUser = ResponseUtil.getAlbumsAndPhotosOfRandomUser(userId, e.albums, 
  18.     e.photos); 
  19.     String response = postsAndCommentsOfRandomUser + albumsAndPhotosOfRandomUser; 
  20.     e.response = response; 
  21.     e.countDownLatch.countDown(); 
  22.     } 
  23.     }).build(); 

11. 總結(jié)

  • 根據(jù)機(jī)器的負(fù)載決定 Executor 框架的配置,并檢查是否可以根據(jù)應(yīng)用程序中并行任務(wù)的數(shù)量進(jìn)行負(fù)載平衡。
  • 對(duì)于大多數(shù)傳統(tǒng)應(yīng)用程序來說,使用響應(yīng)式開發(fā)庫(kù)或任何異步庫(kù)都會(huì)降低性能。只有當(dāng)用例類似于服務(wù)器端聊天應(yīng)用程序時(shí),這個(gè)模式才有用,其中線程在客戶機(jī)響應(yīng)之前不需要保留連接。
  • Disruptor 框架在與事件驅(qū)動(dòng)的架構(gòu)模式一起使用時(shí)性能很好;但是當(dāng) Disruptor 模式與傳統(tǒng)架構(gòu)混合使用時(shí),就我們?cè)谶@里討論的用例而言,它并不符合標(biāo)準(zhǔn)。這里需要注意的是,Akka 和 Disruptor 庫(kù)值得單獨(dú)寫一篇文章,介紹如何使用它們來實(shí)現(xiàn)事件驅(qū)動(dòng)的架構(gòu)模式。
  • 這篇文章的源代碼可以在 GitHub 上找到。

分享文章:鳥瞰Java并發(fā)框架
URL分享:http://www.dlmjj.cn/article/dhchhoj.html