Java 并發編程生產應用場景及實戰-有解無憂

背景介紹

為什么需要學習 Java 并發？

從提升性能角度來說

提升了對CPU的使用效率：目前生產的服務器大多數都是多核，標配的機器都是 8C/16G，作業系統會將不同的執行緒分配給不同的核心處理，理論上，有多少核心就有多少個執行緒并行執行，如果沒有并發編程，CPU的利用率將極大的浪費，假設當前正在處理耗時的 I/O 操作，那么整個CPU就會處于阻塞空閑狀態，后面的指令必須等待前面的執行完才能繼續執行，
降低服務 RT：大型互聯網訪問量輕松每秒輕松過萬，如果沒有并發處理，所有的用戶請求都會排隊等待，那種體驗效果你能想象么，這樣的服務能力如何能留住客戶？有了并發編程，充分釋放CPU算力，作業系統讓每個客戶輪流使用CPU計算，每個客戶都能得到快速的回應，
容錯率高：執行緒與執行緒之間的執行不會相互干擾，某個執行緒執行出現例外退出，不會對其它執行緒造成影響，

從開發者角度來說

Java 基礎面試必考察技能：Java 并發面試問題基本必出現，有大型專案研發經驗的同學，處理并發問題多的同學，往往會被青睞，因為越是復雜的系統，并發請求就越多，簡單的業務 + 并發 = 這個業務不簡單，
作業中離不開并發：多執行緒能充分發揮CPU的計算力，這使得我們不得不了解并發的原理，以免造成執行緒安全問題，給生產帶來損失，常用的中間件中大量運用了并發知識，如 MQ、RPC等，如果不熟悉原理，如何能夠調優中間件的使用，

并發編程業務中的實踐

實踐一：風控規則引擎——策略執行

互聯網企業風控安全部門每時每刻都需要和黑灰產對抗，保護企業遭受不必要的經濟損失，風控策略團隊在對抗的程序中，沉淀出一系列風險識別策略，用以檢測當前業務請求中否存在高危操作，

風控安全團隊需要評估業務在運行流程中，是否存在黑產能夠獲取利益點的地方，即“風險卡點”，評估后，需要業務在每次流程經過風險卡點處，需透傳業務資訊給風控服務，風控服務在很多時間內進行大量決策計算，并回傳業務方決策結果（ACCEPT-通過/REVIEW-人工，需進一步資訊確認/REJECT-拒絕，高危操作），如圖展示的是營銷活動——裂變類活動風險卡點，

營銷裂變流程風險卡點圖

一條業務請求耗時一般在 300 ~ 500 ms 之間，如果超過這個區間，可能就需要定位調優哪個節點耗時高了，大型互聯網公司系統架構比較復雜，完整的業務可能有幾十甚至上百個服務系統，你觸發的一次請求，可能中途會經過多少服務超過你想象，

如上述，業務對風控服務的性能要求很高，一般控制在 100 ms 以內，但風控內部排查業務請求涉及大量策略和規則，如何短時間內執行完，且又不閹割策略呢？答案是并發變成，風控內部大量使用并發，以滿足海量請求和計算需求，我將以策略規則的執行來舉例如何撰寫編發變成代碼，如下是風控服務一次請求的大致執行流程：

風控流程執行圖

可以看到，一次風控請求的判定，涉及大量的規則判定，此時如果沒有并發，會出現什么效果？

串行&并行執行規則圖

全部串行執行策略和規則的話，可能幾秒都不定能計算出來，此時我們需要使用 Java 并發來滿足性能需求，
核心代碼如下：

public class RuleSessionExecutor {

    // 執行緒池
    private final static ExecutorService executor = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors() * 8,
            Runtime.getRuntime().availableProcessors() * 8,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>(),
            new CustomerThreadFactory("rule-executor"),
            new ThreadPoolExecutor.AbortPolicy());

    /**
     * 規則執行
     * @param rules
     */
    public void execute(List<CustomerSession> rules) {
        final CountDownLatch lock = new CountDownLatch(rules.size());
        for (CustomerSession session : rules) {
            try {
                session.exec();
            } catch (Throwable e) {

            } finally {
                lock.countDown();
            }
        }

        try {
            lock.await(50, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            log.error("CountDownLatch error", e);
        }
    }
}

此處用到了 CountDownLatch 并發工具類，下文會使用介紹，

實踐二：風控特征平臺特征加載

如上述實踐一，大量的規則執行前需要大量的特征，如果在每條規則執行內獲取特征，可行，但是會造成特征的重復獲取問題，浪費了性能，舉例：如果規則人員做 A/B 測驗，兩個策略包有交集的特征特別多，此時如果在每個規則內獲取，就等于有交集的特征重復訪問兩次，這種浪費是沒必要的，此時我們在規則執行前先獲取當前策略包下所有的去重特征，然后獲取所有的特征后，再去執行規則，

那么此時的問題是，如何批量的去獲取特征呢？
特征的種類很多:

輸入型：不耗時-請求背景關系攜帶，如訂單金額
衍生型：基本不耗時-基于輸入型特征衍生，如依據經緯度計算距離
實時統計特征：基本不耗時-感興趣的可以關注我文章 Flink 在風控場景實時特征落地實戰，詳細介紹
查詢類：耗時-如依據訂單號調業務 RPC 介面獲取訂單明細資訊，通信 + 業務本身耗時
外部類：耗時-如第三方風控公司產品，同盾、IPIP等

特征同步獲取 & 異步獲取對比圖

顯然，我們需要并發來支撐性能，核心代碼如下：

public class DataSourceExecutor {

    private final static ExecutorService executor = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors() * 8,
            Runtime.getRuntime().availableProcessors() * 8,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>(),
            new HamThreadFactory("ds-executor"),
            new ThreadPoolExecutor.AbortPolicy());

    /**
     * 特征獲取
     *
     * @param dataSources
     */
    public void execute(List<DataSource> dataSources) {
        List<CompletableFuture> futures = Lists.newArrayList();
        long timeout = ApolloConfig.getLongProperty(ApolloConfigKey.DS_TIMEOUT_OUTER_KEY, 150L);

        for (DataSource ds : dataSources) {
            timeout = ds.getExecutionTimeout() > timeout ? ds.getExecutionTimeout() : timeout;

            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                ds.execute();
            }, executor);
            futures.add(future);
        }

        CompletableFuture<Void> summaryFuture = CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[]{}));
        try {
            summaryFuture.get(timeout, TimeUnit.MILLISECONDS);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            log.error("DataSource exec error", e);
        }
    }
}

和規則的批量執行大同小異，單此處用到了 Java 8 CompletableFuture 并發工具類，功能上有所增強，下文會使用介紹，

實踐三：分布式任務跑批

定時任務應該是作業中很常見的需求了，如訂單狀態流轉檢測、對賬等，任務一般都是跑批的，即包含多個子任務，該場景很適合執行緒池任務佇列并發執行，

任務佇列執行緒池圖

核心代碼如下：

public void execute(List<Task> tasks) {
    tasks.forEach(t -> {

        executor.execute(() -> {
            try {
                log.info("task id: {} begin exec", t.getId());
                t.execute();
            } catch (Throwable e) {
                log.error(String.format("task execute error, uid: %s", t.getId()), e);
            } finally {
                log.info("task id: {} end exec", t.getId());
            }
        });

    });
}

并發編程常用工具類

執行緒池

執行緒池（英語：thread pool）：一種執行緒使用模式，執行緒過多會帶來調度開銷，進而影響快取區域性和整體性能，而執行緒池維護著多個執行緒，等待著監督管理者分配可并發執行的任務，這避免了在處理短時間任務時創建與銷毀執行緒的代價，執行緒池不僅能夠保證內核的充分利用，還能防止過分調度，可用執行緒數量應該取決于可用的并發處理器、處理器內核、記憶體、網路sockets等的數量【1】，

J.U.C提供的執行緒池：ThreadPoolExecutor類，幫助開發人員管理執行緒并方便地執行并行任務，了解并合理使用執行緒池，是一個開發人員必修的基本功，

任務調度
當用戶提交了任務，任務的生命周期將有執行緒池管控，執行緒池內部實際上構建了一個生產者/消費者模式，執行緒與任務是解耦的，沒有強關聯性，這有利于任務的緩沖&復用，了解執行緒池的第一步必須知道任務的運行機制，

任務執行圖

任務佇列
執行緒池的本質是對任務和執行緒的管理，而做到這一點的關鍵是解耦任務和執行緒，不讓兩者直接關聯，才能做到后續的合理分配作業，執行緒池中是以生產者消費者模式，通過一個阻塞佇列來實作的，阻塞佇列快取任務，作業執行緒從阻塞佇列中獲取任務，

阻塞佇列（BlockingQueue）在佇列的基礎上新增兩個特性，

佇列為空時，獲取元素的執行緒會等待佇列變為非空
佇列滿時，存盤元素的執行緒會等待佇列可用

阻塞佇列常用于生產者和消費者的場景，生產者是往佇列里添加元素的執行緒，消費者是從佇列里拿元素的執行緒，阻塞佇列就是生產者存放元素的容器，而消費者也只從容器里拿元素，

阻塞佇列示意圖

阻塞佇列如下表可選擇：

ArrayBlockingQueue	有界；陣列實作；FIFO；
LinkedBlockingQueue	有界（默認長度Integer.MAX_VALUE，不小心就會記憶體溢位）；鏈表實作；FIFO；
PriorityBlockingQueue	無界；平衡二叉樹實作；排序
DelayQueue	同PriorityBlockingQueue；物件只能在其到期時才能從佇列中取走
SynchronousQueue	不存盤元素；沒一個put操作需等待take操作
LinkedTransferQueue	有界；優勢在于相對LinkedBlockingQueue降低了鎖的粒度，性能更高
LinkedBlockingDeque	相對LinkedBlockingQueue實作雙端阻塞；鎖粒度降低，性能較好

任務拒絕
執行緒池自我保護熔斷部分，當任務有界快取佇列已滿，證明執行緒池已經超負荷運轉，處理不過來了，此時需要拒絕新進任務，采用設定的拒接策略，以保護執行緒池，

用戶可以選擇JDK提供的四種拒絕策略，或者自定義實作RejectedExecutionHandler介面即可

ThreadPoolExecutor.AbortPolicy()	丟棄任務并拋出RejectedExecutionException例外；執行緒池默認拒絕策略；關鍵業務應使用此例外策略，以了解執行緒池的健康狀況
ThreadPoolExecutor.CallerRunsPolicy()	由主執行緒去執行當前任務
ThreadPoolExecutor.DiscardOldestPolicy()	丟棄老任務，重新提交任務；生產不建議使用，有風險
ThreadPoolExecutor.DiscardPolicy()	丟棄任務&不拋出例外；生產不建議使用，不易發現問題

CountDownLatch——同步計數器

CountDownLatch內部使用計數器實作，初始化時計數器數量等于需要處理的等待執行緒數量，當每個執行緒執行完畢后需要將計數器減一，當計數器到0后，代表需要等待執行的執行緒已全部執行完畢，此時會喚醒主執行緒繼續執行主線任務，

CountDownLatch流程圖

常用場景：

1等N：結合執行緒池釋放CPU算力，如首頁復雜資訊流成精，可以分布加載各模塊資訊，計數結束后在交由主執行緒結構化資料回傳
最壞匹配：哪一個執行緒執行完畢，可以立即釋放cnt數量至0，通知主執行緒執行

核心代碼：

public void demo() {
    List<Task> tasks = ...
    final CountDownLatch countDownLatch = new CountDownLatch(10);
    tasks.forEach(task -> {

        try {
            // 自己的子執行緒邏輯
        } catch (Throwable e) {
            // 有時 Exception 接不到的例外，建議用 Throwable
        } finally {
            countDownLatch.countDown();
        }

    });

    try {
        countDownLatch.await(100, TimeUnit.MILLISECONDS);
    } catch (InterruptedException e) {
        log.error("countDownLatch InterruptedException", e);
    }

}

CompletableFuture

Java 在 1.8 版本提供了 CompletableFuture 來支持異步編程，CompletableFuture 的功能著實讓人感到震撼，他的復雜度應該也是我見過的最復雜之一了，

我們看個例子來直觀感受下 CompletableFurure 的威力

CompletableFuture 之西紅柿炒蛋

// 步驟一：準備西紅柿
CompletableFuture<String> f1 =
  CompletableFuture.supplyAsync(() -> {
    System.out.println("洗西紅柿");

    System.out.println("切它");

    return "切好的西紅柿";
  });

// 步驟二：準備雞蛋
CompletableFuture<String> f2 =
  CompletableFuture.supplyAsync(() -> {
    System.out.println("洗雞蛋");

    System.out.println("煎雞蛋");

    return "煎好的雞蛋";
  });
// 步驟三：炒雞蛋
CompletableFuture<String> f3 =
  f1.thenCombine(f2, (__, tf) -> {
    System.out.println("爆炒");
    return "西紅柿炒雞蛋";
  });

// 等待任務 3 執行結果
System.out.println(f3.join());

CompletableFuture 旨在解決多執行緒之間的復雜實作邏輯，如上所示，其實都是只包含了業務實作的邏輯，并發編程的邏輯已經被 Lamda 編程巧妙的規避了，即用最少的代碼干最硬的事，很完美，

此處不對 CompletableFuture 作詳細的描述，如果感興趣可以關注我，因為 CompletableFuture 實作即使用一篇文章都不一定能說完，

常見并發問題及解決

死鎖&定位

死鎖（deadlock），當兩個以上的運算單元，雙方都在等待對方停止執行，以獲取系統資源，但是沒有一方提前退出時，就稱為死鎖，【1】

死鎖的四個條件是：

禁止搶占（no preemption）：系統資源不能被強制從一個行程中退出，
持有和等待（hold and wait）：一個行程可以在等待時持有系統資源，
互斥（mutual exclusion）：資源只能同時分配給一個行程，無法多個行程共享，
回圈等待（circular waiting）：一系列行程互相持有其他行程所需要的資源，

定位

jps
jstack pid


// 上面的資訊截取

Found one Java-level deadlock:
=============================
"Thread-1":
  waiting to lock monitor 0x00007fcc68023f58 (object 0x0000000795ea0c00, a java.lang.Object),
  which is held by "Thread-0"
"Thread-0":
  waiting to lock monitor 0x00007fcc68022ab8 (object 0x0000000795ea0c10, a java.lang.Object),
  which is held by "Thread-1"

jps 定位運行的 java 程式，然后利用 jstack pid 列印執行緒資訊，拉倒最下面很明顯發現有提示 deadlock，再依據執行緒號 0x00007fcc68023f58 尋找到對應的執行緒即可分析是哪一段代碼引發的問題，

性能調優

Java 并發多執行緒編程，我們首選的工具一定是執行緒池，執行緒池使用面臨的核心的問題在于：執行緒池的引數并不好配置！

你是否會也遇到過按照經驗去預估線上某個場景執行緒池最小活躍執行緒和最大活躍執行緒數不準或失誤，事實上并無執行緒池通用的計算公式，因為一臺機器上并不是只有你的一個服務，且一個服務內并不是只有一個執行緒池，如果按照 I/O密集或者 CPU密集預估，還是免不了反復除錯的苦，

那么我們是否可以將修改執行緒池引數的成本降下來，這樣至少可以發生故障的時候可以快速調整從而縮短故障恢復的時間呢？

本篇不會多講動態執行緒池的架構設計，感興趣的可以關注我，后續會發文，此處只給出一個大概的思路：

動態調參：支持執行緒池核心引數動態調整，最小/最大核心執行緒數
任務監控：主要監控阻塞佇列堆積和單次執行緒執行耗時 95 99 線
告警：有潛在壓力及時預警修正
操作通知&鑒權：生產操作很危險，需謹慎

完善監控

任何系統的運行都離不開監控，只是顆粒度粗細的問題，并發場景，我們尤其需要關注服務執行緒的監控狀況，尤其是活躍執行緒數、佇列堆積長度、平均耗時、吞吐量等重要指標，發生預警時能及時通知相應的開發人員降級處理，亦可自動熔斷，保護主服務，

往期精彩

個人技術博客：https://jifuwei.github.io/
公眾號：是咕咕雞

性能調優——小小的log大大的坑
性能優化必備——火焰圖
Flink 在風控場景實時特征落地實戰

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/508064.html

標籤：其他

上一篇：讀時加寫鎖，寫時加讀鎖，Eureka可真的會玩

下一篇：MyBatis（十）：日志