java中的Executors簡介與多執行緒在網站上逐步優化的運用案例

阿新 • • 發佈：2018-11-26

提供Executor的工廠類
在這裡插入圖片描述

忽略了自定義的ThreadFactory、callable和unconfigurable相關的方法

newFixedxxx:在任意時刻，最多有nThreads個執行緒在處理task；如果所有執行緒都在執行時來了新的任務，它會被扔入佇列；如果有執行緒在執行期間因某種原因終止了執行，如果需要執行後續任務，新的執行緒將取代它
```
   return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
 
```
newCachedxxx:新任務到來如果執行緒池中有空閒的執行緒就複用，否則新建一個執行緒。如果一個執行緒超過60秒沒有使用，它就會被關閉移除執行緒池
```
 return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
```
newSingleThreadExecutor:僅使用一個執行緒來處理任務，如果這執行緒掛了，會產生一個新的執行緒來代替它。每一個任務被保證按照順序執行，而且一次只執行一個
```
  public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
 
```
使用newFixedxxx方法也能實現類似的作用，但是ThreadPoolExecutor會提供修改執行緒數的方法，FinalizableDelegatedExecutorService則沒有修改的途徑，它在DelegatedExecutorService的基礎
上僅提供了執行finalize時候去關閉執行緒，而DelegatedExecutorService僅暴漏ExecutorService自身的方法

newScheduledThreadPool:提供一個執行緒池來延遲或者定期執行任務

  public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
              new DelayedWorkQueue());
    }

newSingleThreadScheduledExecutor:提供單個執行緒來延遲或者定期執行任務，如果執行的執行緒掛了，會生成新的。
```
  return new DelegatedScheduledExecutorService
            (new ScheduledThreadPoolExecutor(1));
```
同樣，它保證返回的Executor自身的執行緒數不可修改

從上述的實現可以看出，核心在於三個部分

ThreadPoolExecutor:提供執行緒數相關的控制
DelegatedExecutorService:僅暴露ExecutorService自身的方法，保證執行緒數不變來實現語義場景
ScheduledExecutorService：提供延遲或者定期執行的功能

對應的，相應也有不同的佇列去實現不同的場景

LinkedBlockingQueue：無界阻塞佇列
SynchronousQueue：沒有消費者消費時，新的任務就會被阻塞
DelayQueue：佇列中的任務過期之後才可以執行，否則無法查詢到佇列中的元素

DelegatedExecutorService

它僅僅是包裝了ExecutorService的方法，交由傳入的ExecutorService來執行，所謂的UnConfigurable實際也就是它沒有暴漏配置各種引數調整的方法

  static class DelegatedExecutorService extends AbstractExecutorService {
        private final ExecutorService e;
        DelegatedExecutorService(ExecutorService executor) { e = executor; }
        public void execute(Runnable command) { e.execute(command); }
        public void shutdown() { e.shutdown(); }
        public List<Runnable> shutdownNow() { return e.shutdownNow(); }
        public boolean isShutdown() { return e.isShutdown(); }
        public boolean isTerminated() { return e.isTerminated(); }
        public boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException {
            return e.awaitTermination(timeout, unit);
        }
        public Future<?> submit(Runnable task) {
            return e.submit(task);
        }
        public <T> Future<T> submit(Callable<T> task) {
            return e.submit(task);
        }
        public <T> Future<T> submit(Runnable task, T result) {
            return e.submit(task, result);
        }
        public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
            throws InterruptedException {
            return e.invokeAll(tasks);
        }
        public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                             long timeout, TimeUnit unit)
            throws InterruptedException {
            return e.invokeAll(tasks, timeout, unit);
        }
        public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
            throws InterruptedException, ExecutionException {
            return e.invokeAny(tasks);
        }
        public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                               long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException {
            return e.invokeAny(tasks, timeout, unit);
        }
    }

ScheduledExecutorService

提供一系列的schedule方法，使得任務可以延遲或者週期性的執行，對應schedule方法會返回ScheduledFuture以供確認是否執行以及是否要取消。它的實現ScheduledThreadPoolExecutor也支援立即執行由submit提交的任務

僅支援相對延遲時間，比如距離現在5分鐘後執行。類似Timer也可以管理延遲任務和週期任務，但是存在一些缺陷：

所有的定時任務只有一個執行緒，如果某個任務執行時間長，將影響其它TimerTask的精確性。ScheduledExecutorService的多執行緒機制可彌補

TimerTask丟擲未檢查的異常，將終止執行緒執行，此時會錯誤的認為任務都取消了。1：可以使用try-catch-finally對相應執行快處理；2：通過execute執行的方法可以設定UncaughtExceptionHandler來接收未捕獲的異常，並作出處理;3:通過submit執行的，將被封裝層ExecutionException重新丟擲

ThreadPoolExecutor

 public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

corePoolSize、maximumPoolSize：ThreadPoolExecutor會根據這兩自動調整執行緒池的大小，當一個新任務通過execute提交的時候：
如果當前執行的執行緒數小於corePoolSize就新建執行緒；
如果當前執行緒數在corePoolSize與maximumPoolSize之間，則只有在佇列滿的時候才會建立新的執行緒；
如果已經達到最大執行緒數，並且佇列都滿了，在這種飽和狀態下就會執行拒絕策略
預設情況下，只有新任務到達的時候才會啟動執行緒，可通過prestartCoreThread方法實現事先啟動
1. corePoolSize：預設執行緒池所需要維護的最小的worker的數量，就算是worker過期了也會保留。如果想要不保留，則需要設定allowCoreThreadTimeOut，此時最小的就是0
2. maximumPoolSize：執行緒池最大的執行緒數。java限制最多為 2^29-1,大約5億個
keepAliveTime、unit：如果當前執行緒池有超過corePoolSize的執行緒數，只要有執行緒空閒時間超過keepAliveTime的設定，就會被終止；unit則是它的時間單位
workQueue：任何BlockingQueue都可以使用，基本上有三種
1. Direct handoffs，直接交付任務。比如 SynchronousQueue，如果沒有執行緒消費，提交任務會失敗，當然可以新建一個執行緒來處理。它適合處理有依賴關係的任務，一般它的maximumPoolSizes會被設定成最大的
2. Unbounded queues，無界佇列。比如LinkedBlockingQueue，這意味著如果有corePoolSize個執行緒在執行，那麼其他的任務都只能等待。它適合於處理任務都是互相獨立的，
3. Bounded queues，有界佇列。比如ArrayBlockingQueue，需要考慮佇列大小和最大執行緒數之間的關係，來達到更好的資源利用率和吞吐量
threadFactory：沒有指定的時候，使用Executors.defaultThreadFactory
RejectedExecutionHandler：通過execute新增的任務，如果Executor已經關閉或者已經飽和了(執行緒數達到了maximumPoolSize,並且佇列滿了)，就會執行，java提供了4種策略：
1. AbortPolicy，拒絕的時候丟擲執行時異常RejectedExecutionException；
2. CallerRunsPolicy，如果executor沒有關閉，那麼由呼叫execute的執行緒來執行它；
3. DiscardPolicy，直接扔掉新的任務；
4. DiscardOldestPolicy，如果executor沒有關閉，那麼扔掉佇列頭部的任務，再次嘗試；

ThreadPoolExecutor可自定義beforeExecutor、afterExecutor可以用來新增日誌統計、計時、件事或統計資訊收集功能，無論run是正常返回還是丟擲異常，afterExecutor都會被執行。如果beforeExecutor丟擲RuntimeException,任務和afterExecutor都不會被執行。terminated在所有任務都已經完成，並且所有工作者執行緒關閉後會呼叫，此時也可以用來執行傳送通知、記錄日誌等等。

如何估算執行緒池的大小

計算密集型，通常在擁有 $N_{cpu}$ 個處理器的系統上，執行緒池大小設定為 $N_{cpu}+1$ 能夠實現最優的利用率；

$N_{cpu}$ cpu的個數
I/O密集型或者其它阻塞型的任務，定義 $N_{cpu}$ 為CPU的個數， $U_{cpu}$ 為CPU的利用率， $W/C$ 為等待時間與計算時間的比率，此時執行緒池的最優大小為
$N_{threads}=N_{cpu}*U_{cpu}*(1+W/C)$

場景說明

將一個網站的業務抽象成如下幾塊

接收客戶端請求與處理請求
頁面渲染返回的文字和圖片
獲取頁面的廣告

接收請求與處理請求

理論模型

理論上，服務端通過實現約定的介面就可以實現接收請求和處理連續不斷的請求過來

ServerSocket socket = new ServerSocket(80);
while(true){
    Socket conn = socket.accept();
    handleRequest(conn)
}

缺點：每次只能處理一個請求，新請求到來時，必須等到正在處理的請求處理完成，才能接收新的請求

顯示的建立多執行緒

為每個請求建立新的執行緒提供服務

ServerSocket socket = new ServerSocket(80);
while(true){
    final Socket conn = socket.accept();
    Runnable task = new Runnable(){
        public void run(){
            handleRequest(conn);        
        }
    }
    new Thread(task).start();
}

缺點：

執行緒的建立和銷燬都有一定的開銷，延遲對請求的處理；
建立後的執行緒多於可用處理器的數量，造成執行緒閒置，這會給垃圾回收帶來壓力
存活的大量執行緒競爭CPU資源會產生很多效能開銷
系統上對可建立的執行緒數存在限制

使用執行緒池

使用java自帶的Executor框架。

private static final Executor exec = Executors.newFixedThreadPool(100);
...
ServerSocket socket = new ServerSocket(80);
while(true){
    final Socket conn = socket.accept();
    Runnable task = new Runnable(){
        public void run(){
            handleRequest(conn);        
        }
    }
    exec.execute(task);
}
...

執行緒池策略通過實現預估好的執行緒需求，限制併發任務的數量，重用現有的執行緒，解決每次建立執行緒的資源耗盡、競爭過於激烈和頻繁建立的問題，也囊括了執行緒的優勢，解耦了任務提交和任務執行。

頁面渲染返回的文字和圖片

序列渲染

renderText(source);
List<ImageData> imageData = new ArrayList<ImageData>();
for(ImageInfo info:scaForImageInfo(source)){
    imageData.add(info.downloadImage());
}
for(ImageData data:imageData){
    renderImage(data);
}

缺點：影象的下載大部分時間在等待I/O操作執行完成，這期間CPU幾乎不做任何工作，使得使用者看到最終頁面之前要等待過長的時間

並行化

渲染過程可以分成兩個部分，1是渲染文字，1是下載影象

private static final ExecutorService exec = Executors.newFixedThreadPool(100);
...
final List<ImageInfo> infos=scaForImageInfo(source);
Callable<List<ImageData>> task=new Callable<List<ImageData>>(){
    public List<ImageData> call(){
        List<ImageData> r = new ArrayList<ImageData>();
        for(ImageInfo info:infos){
            r.add(info.downloadImage());
        }
        return r;
    }
};
Future<List<ImageData>> future = exec.submit(task);
renderText(source);
try{
    List<ImageData> imageData = future.get();
    for(ImageData data:imageData){
        renderImage(data);
    }
}catch(InterruptedException e){
    Thread.currentThread().interrupt();
    future.cancel(true);
}catche(ExecutionException e){
    throw launderThrowable(e.getCause());
}

使用Callable來返回下載的圖片結果，使用future來獲得下載的圖片，這樣將減少使用者所需要的等待時間。
缺點：圖片的下載很明顯時間要比文字要慢，這樣的並行化很可能速度可能只提升了1%

並行效能提升

使用CompletionService。

private static final ExecutorService exec;
...
final List<ImageInfo> infos=scaForImageInfo(source);
CompletionService<ImageData> cService =  new ExecutorCompletionService<ImageData>(exec);
for(final ImageInfo info:infos){
    cService.submit(new Callable<ImageData>(){
        public ImageData call(){
            return info.downloadImage();
        }
    });
}
renderText(source);
try{
    for(int i=0,n=info.size();t<n;t++){
        Future<ImageData> f = cService.take();
        ImageData imageData=f.get();
        renderImage(imageData)
    }
}catch(InterruptedException e){
    Thread.currentThread().interrupt();
}catche(ExecutionException e){
    throw launderThrowable(e.getCause());
}

核心思路為為每一幅影象下載都建立一個獨立的任務，並在執行緒池中執行他們，從而將序列的下載過程轉換為並行的過程

獲取頁面的廣告

廣告展示如果在一定的時間以內沒有獲取，可以不再展示，並取消超時的任務。

 ExecutorService exe = Executors.newFixedThreadPool(3);
        List<MyTask> myTasks = new ArrayList<>();
        for (int i=0;i<3;i++){
          myTasks.add(new MyTask(3-i));
        }

        try {

            List<Future<String>> futures = exe.invokeAll(myTasks, 1, TimeUnit.SECONDS);
            for (int i=0;i<futures.size();i++){
                try {
                    String s = futures.get(i).get();
                    System.out.println("task execut "+myTasks.get(i).getSleepSeconds()+" s");
                } catch (ExecutionException e) {
                    System.out.println("task sleep "+myTasks.get(i).getSleepSeconds()+" not execute ");
                }catch (CancellationException e){
                    System.out.println("task sleep "+myTasks.get(i).getSleepSeconds()+" not execute ,because "+e);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        exe.shutdown();

invokeAll方法對於沒有完成的任務會被取消，通過CancellationException可以捕獲，invokeAll返回的序列順序和傳入的task保持一致。結果如下：

task sleep 3 not execute ,because java.util.concurrent.CancellationException
task sleep 2 not execute ,because java.util.concurrent.CancellationException
task execut 1 s

java中的Executors簡介與多執行緒在網站上逐步優化的運用案例

提供Executor的工廠類忽略了自定義的ThreadFactory、callable和unconfigurable相關的方法 newFixedxxx:在任意時刻，最多有nThreads個執行緒在處理task；如果所有執行緒都在執行時來了新的任務，它會被扔

從專案中談JAVA中static 方法在多執行緒環境下的運用

最近在做一個實時性比較高的系統，資料庫訪問很頻繁。在這個專案中資料庫訪問的大多數都被我做成了static 方法。(這樣做應該不是很好，沒辦法發揮dao模式的優勢，不過湊合用吧)因這個系統是多執行緒。static方法的運用需要注意，不能在servlet中使用成員變數，因為如果你

Java中使用CountDownLatch進行多執行緒同步

CountDownLatch介紹在前面的Java學習筆記中，總結了Java中進行多執行緒同步的幾個方法：1、synchronized關鍵字進行同步。2、Lock鎖介面及其實現類ReentrantLock

java中為什麼要用多執行緒

我們可以在計算機上執行各種計算機軟體程式。每一個執行的程式可能包括多個獨立執行的執行緒（Thread）。執行緒（Thread）是一份獨立執行的程式，有自己專用的執行棧。執行緒有可能和其他執行緒共享一些資源，比如，記憶體，檔案，資料庫等。當多個執行緒同時讀寫同一份共享資源

JAVA中基於UDP實現多執行緒通訊

伺服器端程式，利用DatagramSocket負責監聽埠，當客戶端發過來訊息時，伺服器端就會響應，並將訊息內容儲存到Datagrampacket物件中，並且！每一次while迴圈必須重新建立DatagramPacket物件用於儲存訊息資料。並將socket，pa

java面試/筆試題目之多執行緒及鎖 (持續更新中)

前言：這一模組可以參照徐劉根大佬的部落格。一.執行緒和程序的概念、並行和併發的概念 1.程序：是計算機中的程式關於某資料集合上的一次執行活動，是系統進行資源分配和排程的基本單位，是作業系統結構的基礎。程式是指令、資料及其組織形式的描述，程序是程式的實體。 2.執行緒：是程式執行流的

c#中委託與多執行緒的實質

delegate（委託）的概念，.Net的委託本質上就是指向函式的指標，只不過這種指標是經過封裝後型別安全的。委託和執行緒是兩個不同的概念，執行緒是動態的，委託就是一個或一組記憶體地址，是靜態的。執行緒執行時如果遇到了指向函式的指標就執行這個函式。.Net為了方便程式設計，給委託賦予了兩種方式以供呼

Java多執行緒學習筆記21之單例模式與多執行緒

詳細程式碼見:github程式碼地址第六章單例模式與多執行緒前言: 我之前已經開設了23個設計模式這個專欄，介紹了很多的Java設計模式，其中一些模式對於絕大多數程式語言設計思想都是類似的，需要了解單例模式的可以去看看。我們在實際開發中經常用到單例模式，但

Java執行緒安全與多執行緒開發

網際網路上充斥著對Java多執行緒程式設計的介紹，每篇文章都從不同的角度介紹並總結了該領域的內容。但大部分文章都沒有說明多執行緒的實現本質，沒能讓開發者真正“過癮”。從Java的執行緒安全鼻祖內建鎖介紹開始，讓你瞭解內建鎖的實現邏輯和原理以及引發的效能問題，接著說明了Java多執行緒程式設計中鎖的存在是為

[Xcode10 實際操作]八、網路與多執行緒-(15)使用網址會話物件URLSession下載圖片並存儲在沙箱目錄中

本文將演示如何通過網址會話物件URLSession下載圖片並存儲在沙箱目錄中。網址會話物件URLSession具有在後臺上傳和下載、暫停和恢復網路操作、豐富的代理模式等優點。在專案導航區，開啟檢視控制器的程式碼檔案【ViewController.swift】 1 import UIKit

一位10年Java程式設計師總結進階中的你懂多執行緒和jvm優化嗎？

感謝朋友們的認可和指正。本文是有感而發，因為看過了太多坑人的部落格和書籍，感慨自己走過的彎路，不希望其他初學者被網上互相抄襲的部落格和東拼西湊的書籍浪費時間，想以一個相對巨集觀的視野來描述一個概念，力求通俗易懂，所以沒有深入太多細節，簡化了很多模型，給部分朋友造成了疑惑，說聲抱歉。也沒有配圖，都是抽

Java多執行緒核心技術(五)單例模式與多執行緒

本文只需要考慮一件事：如何使單例模式遇到多執行緒是安全的、正確的 1.立即載入 / "餓漢模式" 什麼是立即載入？立即載入就是使用類的時候已經將物件建立完畢，常見的實現辦法就是直接 new 例項化。 public class MyObject { private static MyObject m

Java面向物件與多執行緒綜合實驗（五）之JDBC

本次程式碼沿用第三次中的User.java Administrator.java Operator.java Browser.java和Doc.java 以及第四次中的LoginWindow.java MenuWindow.java UserWindow.java UpDownloadWin

Java面向物件與多執行緒綜合實驗（四）之GUI設計

瞭解Java圖形介面程式的基本結構；掌握Java佈局管理和常用元件的使用；掌握Java事件處理機制。實驗內容編寫程式，將前面課程所編寫的檔案管理系統改編為圖形使用者介面。要求程式介面選用合適的佈局，綜合使用選單、按鈕、文字框、密碼框、下拉列表、檔案對話方塊等元件，實現良好的人機介面。

Java面向物件與多執行緒綜合實驗（三）之輸入輸出流

瞭解Java中I/O流的概念和種類；掌握位元組流處理和字元流處理，包括File類，InputStream/OutputStream及其子類，Reader/Writer及其子類；熟練掌握檔案的順序處理，隨機訪問處理；熟悉物件序列化的概念和方法。編寫程式，實現檔案管理系統中的檔案上傳/下載模組

Java面向物件與多執行緒綜合實驗（二）之異常處理

理解異常的基本概念；瞭解Java異常的層次結構；熟悉並掌握Java異常的捕獲處理方法。（1）閱讀Java™ Platform, Standard Edition 8 API Specification文件，瞭解後續程式設計中將要處理的IOException及其子類FileNotFoundE

Java面向物件與多執行緒綜合實驗（一）之封裝、繼承與多型

編寫一個程式，實現檔案管理系統中的使用者管理模組。要求模組中實現使用者的模擬登入過程。通過使用者輸入，獲取使用者名稱和口令；與事先記錄在程式中的使用者資訊進行對比，通過口令驗證後才能使用系統。使用者分為系統管理人員、檔案錄入人員，檔案瀏覽人員三類，相關類圖如下所示。（1）要求在使用者類中

MySQL---當Java遇上MySQL⑤---單執行緒與多執行緒下的事務

事務transaction 原子性(atomicity)：組成事務處理的語句形成了一個邏輯單元，不能只執行其中的一部分。一致性(consistency)：在事務處理執行前後，資料庫是一致的(資料庫資料完整性約束)。隔離性(isolcation)：一個事務處理對另

Java多執行緒的實現（程序與執行緒的概念、Java繼承Thread類實現多執行緒、Java實現Runnable介面實現多執行緒、Thread與Runnable的區別、實現Callable介面實現多執行緒）

1 程序與執行緒 1.1 程序與執行緒的概念什麼是程序？程序：作業系統中一個程式的執行週期。（比如我們想要在電腦上登入QQ，從雙擊qq按鈕---->關閉qq這個過程就是一個程序）多程序：同一時刻跑多個程式。在DOS（磁碟作業系統時

【Java 單例模式】Java 單例模式在多執行緒環境中可能存在的問題

在多執行緒環境下，使用延遲載入的方式實現單例模式，會出現錯誤。例如，使用如下方式實現單例類： package study20170307; /** * Created by apple on 17/3/7. */ public class Sin

java中的Executors簡介與多執行緒在網站上逐步優化的運用案例

DelegatedExecutorService

ScheduledExecutorService

ThreadPoolExecutor

如何估算執行緒池的大小

場景說明

接收請求與處理請求

理論模型

顯示的建立多執行緒

使用執行緒池

頁面渲染返回的文字和圖片

序列渲染

並行化

並行效能提升

獲取頁面的廣告

相關推薦