1. 從“并發(fā)”到“高并發(fā)”一線工程師的實(shí)戰(zhàn)視角“多線程”和“高并發(fā)”這兩個(gè)詞在Java工程師的日常里就像空氣和水一樣常見但真正能把它們玩明白、玩出花來的卻不多。很多朋友學(xué)了一堆synchronized、volatile背了幾道面試題一上生產(chǎn)環(huán)境面對(duì)真實(shí)的流量洪峰系統(tǒng)該崩還是崩。問題出在哪在我看來是缺少一套從“玩具代碼”到“工業(yè)級(jí)應(yīng)用”的完整知識(shí)體系和實(shí)戰(zhàn)心法。JUCjava.util.concurrent包就是這套心法的核心秘籍它遠(yuǎn)不止是幾個(gè)Atomic類或者ConcurrentHashMap那么簡(jiǎn)單而是一整套應(yīng)對(duì)高并發(fā)復(fù)雜場(chǎng)景的“工具箱”和“設(shè)計(jì)模式”。我經(jīng)歷過不少從零到一搭建高并發(fā)服務(wù)的項(xiàng)目也處理過不少因?yàn)椴l(fā)問題導(dǎo)致的線上事故。今天我們不聊那些枯燥的概念就從一個(gè)一線工程師的視角掰開揉碎了講講在真實(shí)的“大廠”級(jí)場(chǎng)景下JUC里的那些工具到底該怎么用背后的“為什么”又是什么。目標(biāo)很明確讓你不僅能應(yīng)對(duì)面試更能寫出健壯、高效、易于維護(hù)的并發(fā)代碼。學(xué)妹收藏不收藏不重要重要的是這些經(jīng)驗(yàn)?zāi)苷嬲龓湍阍陧?xiàng)目里少踩坑。2. JUC核心工具箱不止于API更是設(shè)計(jì)思想很多人學(xué)JUC喜歡從一個(gè)個(gè)類開始背方法。這就像學(xué)武功只記招式不懂內(nèi)功心法實(shí)戰(zhàn)起來必然僵硬。我認(rèn)為理解JUC首先要建立三層認(rèn)知同步工具、并發(fā)容器和執(zhí)行框架。這三層環(huán)環(huán)相扣構(gòu)成了處理并發(fā)問題的完整體系。2.1 同步工具從“鎖”到“協(xié)作”的進(jìn)化synchronized和ReentrantLock解決了互斥問題但這只是并發(fā)世界最基礎(chǔ)的一環(huán)。在高并發(fā)場(chǎng)景中線程間的“協(xié)作”往往比單純的“互斥”更復(fù)雜、也更容易出問題。CountDownLatch多線程任務(wù)的“發(fā)令槍”想象一個(gè)電商系統(tǒng)啟動(dòng)時(shí)需要加載緩存、初始化連接池、校驗(yàn)配置文件等多個(gè)前置任務(wù)。這些任務(wù)可以并行執(zhí)行但必須全部完成后服務(wù)才能對(duì)外提供。用Thread.join()或者忙等待while循環(huán)都太笨重了。// 實(shí)戰(zhàn)場(chǎng)景服務(wù)啟動(dòng)同步 public class ServiceBootstrap { private static final int TASK_COUNT 3; private final CountDownLatch latch new CountDownLatch(TASK_COUNT); public void start() throws InterruptedException { ExecutorService executor Executors.newFixedThreadPool(TASK_COUNT); // 并行執(zhí)行初始化任務(wù) executor.submit(() - { try { loadCache(); // 模擬耗時(shí)操作 } finally { latch.countDown(); } }); executor.submit(() - { try { initConnectionPool(); } finally { latch.countDown(); } }); executor.submit(() - { try { validateConfig(); } finally { latch.countDown(); } }); // 等待所有前置任務(wù)完成設(shè)置超時(shí)避免死等 if (latch.await(30, TimeUnit.SECONDS)) { System.out.println(所有服務(wù)初始化完成開始接收外部請(qǐng)求。); } else { System.err.println(服務(wù)初始化超時(shí)可能存在異常); // 這里應(yīng)該觸發(fā)優(yōu)雅降級(jí)或告警 } executor.shutdown(); } }注意countDown()一定要放在finally塊中執(zhí)行確保無論任務(wù)成功與否計(jì)數(shù)器都能遞減防止主線程永遠(yuǎn)等待。超時(shí)設(shè)置是生產(chǎn)環(huán)境的必備項(xiàng)絕不能少。CyclicBarriervsCountDownLatch可重復(fù)使用的“集合點(diǎn)”CountDownLatch是一次性的計(jì)數(shù)器減到零就失效。而CyclicBarrier是可循環(huán)使用的它更像一個(gè)“集合點(diǎn)”。一個(gè)經(jīng)典的應(yīng)用場(chǎng)景是數(shù)據(jù)分片計(jì)算將一個(gè)大任務(wù)拆分成多個(gè)子任務(wù)并行處理所有子任務(wù)都完成一個(gè)階段后再一起進(jìn)入下一個(gè)階段。// 模擬多階段數(shù)據(jù)批處理 public class BatchDataProcessor { private final int workerCount; private final CyclicBarrier barrier; public BatchDataProcessor(int workerCount) { this.workerCount workerCount; // 當(dāng)所有線程到達(dá)屏障后可以選擇執(zhí)行一個(gè)回調(diào)Runnable用于合并階段結(jié)果 this.barrier new CyclicBarrier(workerCount, () - { System.out.println(所有分片第一階段處理完成開始匯總...); // 這里可以執(zhí)行階段性的數(shù)據(jù)聚合操作 }); } public void process(ListDataSlice slices) { ExecutorService executor Executors.newFixedThreadPool(workerCount); for (int i 0; i workerCount; i) { final int sliceIndex i; executor.submit(() - { try { // 第一階段處理 phaseOneProcess(slices.get(sliceIndex)); barrier.await(); // 等待其他線程完成第一階段 // 第二階段處理基于第一階段可能匯總的結(jié)果 phaseTwoProcess(slices.get(sliceIndex)); barrier.await(); } catch (Exception e) { Thread.currentThread().interrupt(); } }); } executor.shutdown(); } }實(shí)操心得CyclicBarrier的構(gòu)造器中的Runnable回調(diào)是由最后一個(gè)到達(dá)屏障的線程執(zhí)行的且在執(zhí)行期間其他線程仍處于等待狀態(tài)。這個(gè)回調(diào)不宜有耗時(shí)或阻塞操作否則會(huì)影響整體性能。Semaphore控制并發(fā)訪問的“流量閥”信號(hào)量用來控制同時(shí)訪問特定資源的線程數(shù)量。它最典型的應(yīng)用場(chǎng)景就是資源池管理如數(shù)據(jù)庫連接池和限流。// 實(shí)現(xiàn)一個(gè)簡(jiǎn)單的連接池 public class SimpleConnectionPool { private final LinkedListConnection pool new LinkedList(); private final Semaphore useful; public SimpleConnectionPool(int size) { this.useful new Semaphore(size); for (int i 0; i size; i) { pool.addLast(createConnection()); } } public Connection getConnection() throws InterruptedException { useful.acquire(); // 獲取一個(gè)許可如果沒有則阻塞 synchronized (pool) { return pool.removeFirst(); } } public void releaseConnection(Connection conn) { synchronized (pool) { pool.addLast(conn); } useful.release(); // 釋放一個(gè)許可 } }避坑指南務(wù)必保證release()方法一定會(huì)被調(diào)用通常需要放在finally塊中。否則許可無法歸還最終會(huì)導(dǎo)致所有線程都無法獲取資源造成“假死”。在Spring管理的項(xiàng)目中可以利用Around注解的切面來確保資源釋放。2.2 并發(fā)容器告別手動(dòng)同步的“性能陷阱”Hashtable和用Collections.synchronizedMap包裝的HashMap其同步粒度是整個(gè)對(duì)象每次只有一個(gè)線程能進(jìn)行操作性能是巨大的瓶頸。JUC提供的并發(fā)容器采用了更精妙的并發(fā)控制策略。ConcurrentHashMap分段鎖與CAS的藝術(shù)這是面試高頻點(diǎn)也是實(shí)戰(zhàn)核心。在JDK 1.7及之前它采用分段鎖Segment將數(shù)據(jù)分成一段一段的存儲(chǔ)每段配一把鎖不同段的操作可以并發(fā)。在JDK 1.8之后它做了巨大優(yōu)化摒棄了分段鎖改用Node數(shù)組鏈表/紅黑樹并發(fā)控制則大量使用了synchronized和CASCompare-And-Swap操作。關(guān)鍵方法putVal的并發(fā)邏輯當(dāng)要向一個(gè)空桶數(shù)組位置插入節(jié)點(diǎn)時(shí)使用CAS操作避免加鎖。只有當(dāng)發(fā)生哈希沖突桶非空時(shí)才使用synchronized鎖住這個(gè)桶的頭節(jié)點(diǎn)。這種細(xì)粒度的鎖大大提升了并發(fā)度。size()方法的變化1.7版本需要全局加鎖或分段統(tǒng)計(jì)比較重。1.8版本采用了一個(gè)volatile的baseCount變量結(jié)合CounterCell數(shù)組一種分片計(jì)數(shù)思想通過累加來獲取一個(gè)估計(jì)值性能極高且是弱一致性的這符合并發(fā)場(chǎng)景的常態(tài)。重要認(rèn)知ConcurrentHashMap提供的迭代器是“弱一致性”的它反映的是創(chuàng)建迭代器那一刻或之后某個(gè)時(shí)刻的映射狀態(tài)但不會(huì)拋出ConcurrentModificationException。這意味著在迭代過程中其他線程的修改可能看到也可能看不到。這在并發(fā)環(huán)境下是合理的因?yàn)閺?qiáng)一致性的迭代器需要全局鎖代價(jià)太高。CopyOnWriteArrayList讀多寫少場(chǎng)景的“利器”它的原理是“寫時(shí)復(fù)制”。任何修改操作add, set, remove都會(huì)底層復(fù)制一個(gè)新的數(shù)組在新數(shù)組上操作完成后再將原數(shù)組引用指向新數(shù)組。這種機(jī)制使得讀操作完全無需加鎖速度極快。// 典型場(chǎng)景監(jiān)聽器列表 public class EventManager { private final CopyOnWriteArrayListEventListener listeners new CopyOnWriteArrayList(); public void addListener(EventListener listener) { listeners.add(listener); // 寫操作會(huì)復(fù)制數(shù)組 } public void fireEvent(Event event) { for (EventListener listener : listeners) { // 讀操作無鎖直接遍歷當(dāng)前數(shù)組快照 listener.onEvent(event); } } }使用限制它只適用于讀操作遠(yuǎn)遠(yuǎn)多于寫操作的場(chǎng)景。因?yàn)槊看螌懖僮鞫紩?huì)復(fù)制整個(gè)底層數(shù)組如果數(shù)組很大或?qū)懖僮黝l繁內(nèi)存和CPU開銷會(huì)非常大。同時(shí)它提供的迭代器也是基于創(chuàng)建時(shí)的數(shù)組快照無法感知后續(xù)的修改。阻塞隊(duì)列生產(chǎn)者-消費(fèi)者模式的“標(biāo)準(zhǔn)實(shí)現(xiàn)”BlockingQueue及其實(shí)現(xiàn)類ArrayBlockingQueue,LinkedBlockingQueue,PriorityBlockingQueue,SynchronousQueue等是解耦生產(chǎn)者和消費(fèi)者的最佳實(shí)踐。它們內(nèi)部實(shí)現(xiàn)了完整的等待/通知機(jī)制我們無需再手動(dòng)wait()和notify()。ArrayBlockingQueuevsLinkedBlockingQueue特性ArrayBlockingQueueLinkedBlockingQueue底層結(jié)構(gòu)定長(zhǎng)數(shù)組可選容量的鏈表默認(rèn)Integer.MAX_VALUE鎖分離一把鎖生產(chǎn)消費(fèi)共用兩把鎖putLock和takeLock適用場(chǎng)景固定大小的有界隊(duì)列吞吐量預(yù)測(cè)穩(wěn)定無界或可有界高并發(fā)下吞吐量通常更高SynchronousQueue一個(gè)“手遞手”的隊(duì)列。它不存儲(chǔ)元素每個(gè)插入操作必須等待另一個(gè)線程的移除操作反之亦然。它直接傳遞任務(wù)避免了任務(wù)在隊(duì)列中的中轉(zhuǎn)延遲是Executors.newCachedThreadPool默認(rèn)使用的隊(duì)列非常適合大量短生命周期的異步任務(wù)。2.3 原子類無鎖編程的“基石”AtomicInteger、AtomicLong、AtomicReference等原子類是CAS操作的直接體現(xiàn)。它們通過Unsafe類調(diào)用CPU底層的原子指令如x86的CMPXCHG實(shí)現(xiàn)了非阻塞的線程安全更新。// 一個(gè)常見的誤區(qū)原子類并不保證復(fù)合操作的原子性 public class AtomicMisuseExample { private final AtomicInteger count new AtomicInteger(0); // 這個(gè)方法不是線程安全的 public void unsafeIncrement() { if (count.get() 10) { // 步驟1檢查 count.incrementAndGet(); // 步驟2遞增 } // 問題線程A和B可能同時(shí)通過步驟1的檢查導(dǎo)致最終count超過10。 } // 正確的做法使用CAS循環(huán) public void safeIncrement() { int oldValue; do { oldValue count.get(); if (oldValue 10) { return; // 或拋出異常 } } while (!count.compareAndSet(oldValue, oldValue 1)); // CAS更新 } }核心原理compareAndSetCAS是一個(gè)“比較并交換”的原子操作。它的語義是“如果當(dāng)前值等于期望值oldValue則將其更新為新值否則什么都不做”。上面的循環(huán)會(huì)不斷重試直到成功更新或條件不滿足。這就是無鎖Lock-Free編程的一種常見模式。LongAdder高并發(fā)統(tǒng)計(jì)的“性能王者”在超高并發(fā)比如統(tǒng)計(jì)接口調(diào)用次數(shù)的場(chǎng)景下所有線程都去競(jìng)爭(zhēng)更新一個(gè)AtomicLong的valueCAS失敗重試會(huì)非常頻繁導(dǎo)致性能下降。LongAdder采用了“分治”思想。它內(nèi)部維護(hù)了一個(gè)Cell數(shù)組每個(gè)Cell是一個(gè)AtomicLong和一個(gè)base值。當(dāng)沒有競(jìng)爭(zhēng)時(shí)直接CAS更新base。當(dāng)發(fā)生競(jìng)爭(zhēng)時(shí)線程會(huì)嘗試操作自己哈希到的那個(gè)Cell將競(jìng)爭(zhēng)分散。獲取最終結(jié)果時(shí)將base和所有Cell的值累加。 這樣在高并發(fā)寫場(chǎng)景下LongAdder的吞吐量遠(yuǎn)高于AtomicLong但缺點(diǎn)是獲取當(dāng)前值的開銷稍大且是最終一致性的。它非常適合用于統(tǒng)計(jì)、計(jì)數(shù)的場(chǎng)景而不適合用于需要實(shí)時(shí)精確值的場(chǎng)景如序列號(hào)生成。3.ThreadPoolExecutor你必須親手“調(diào)教”的并發(fā)引擎Executors工廠類提供的newFixedThreadPool、newCachedThreadPool等快捷方法在簡(jiǎn)單 demo 里用用可以但在生產(chǎn)環(huán)境直接使用無異于埋雷。它們隱藏了關(guān)鍵的參數(shù)配置容易導(dǎo)致OOM內(nèi)存溢出或資源耗盡。我們必須掌握ThreadPoolExecutor的七大核心參數(shù)并理解其工作原理。3.1 七大核心參數(shù)深度解析public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueueRunnable workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)corePoolSize核心線程數(shù)線程池的“常備軍”。即使它們空閑除非設(shè)置了allowCoreThreadTimeOut否則不會(huì)被回收。這個(gè)值應(yīng)根據(jù)任務(wù)類型CPU密集型/IO密集型和機(jī)器核心數(shù)來設(shè)定。一個(gè)經(jīng)驗(yàn)公式CPU密集型任務(wù)corePoolSize CPU核數(shù) 1IO密集型任務(wù)corePoolSize CPU核數(shù) * 2。但這只是起點(diǎn)必須通過壓測(cè)調(diào)整。maximumPoolSize最大線程數(shù)線程池的“總兵力上限”。當(dāng)工作隊(duì)列滿了且核心線程都在忙線程池會(huì)創(chuàng)建新線程救火隊(duì)直到達(dá)到此上限。keepAliveTimeunit空閑線程存活時(shí)間超過核心線程數(shù)的那些“救火隊(duì)員”如果空閑時(shí)間超過這個(gè)值就會(huì)被回收以節(jié)省資源。workQueue工作隊(duì)列任務(wù)的“緩沖區(qū)”。這是調(diào)優(yōu)的關(guān)鍵也是容易出問題的地方。常見的隊(duì)列選擇策略LinkedBlockingQueue無界隊(duì)列newFixedThreadPool使用它。任務(wù)可以無限堆積直到耗盡內(nèi)存。最大線程數(shù)參數(shù)將失效因?yàn)殛?duì)列永遠(yuǎn)不會(huì)滿不會(huì)觸發(fā)創(chuàng)建新線程。SynchronousQueue直接傳遞隊(duì)列newCachedThreadPool使用它。它沒有容量來一個(gè)任務(wù)如果沒有空閑線程就必須創(chuàng)建新線程執(zhí)行。這會(huì)導(dǎo)致線程數(shù)暴增可能耗盡資源。ArrayBlockingQueue有界隊(duì)列這是生產(chǎn)環(huán)境更推薦的方式。它結(jié)合了核心線程、有界隊(duì)列和最大線程數(shù)形成了穩(wěn)定的處理模型。threadFactory線程工廠用于創(chuàng)建新線程。強(qiáng)烈建議自定義以便為線程設(shè)置有意義的名字如order-process-thread-%d、設(shè)置為守護(hù)線程、或指定異常處理器。這在排查問題時(shí)通過線程名就能快速定位價(jià)值巨大。public class NamedThreadFactory implements ThreadFactory { private final AtomicInteger threadNumber new AtomicInteger(1); private final String namePrefix; NamedThreadFactory(String poolName) { namePrefix poolName -thread-; } public Thread newThread(Runnable r) { Thread t new Thread(r, namePrefix threadNumber.getAndIncrement()); t.setDaemon(false); // 通常設(shè)置為非守護(hù)線程 t.setUncaughtExceptionHandler((thread, throwable) - { // 在這里記錄線程池內(nèi)未捕獲的異常非常重要 System.err.println(Uncaught exception in pool thread: thread.getName(), throwable); }); return t; } }handler拒絕策略當(dāng)線程池已關(guān)閉或隊(duì)列已滿且線程數(shù)達(dá)到最大值時(shí)新提交的任務(wù)該如何處理。JDK提供了四種內(nèi)置策略AbortPolicy默認(rèn)直接拋出RejectedExecutionException。這是最直接的方式讓調(diào)用者感知到系統(tǒng)已過載。CallerRunsPolicy讓提交任務(wù)的調(diào)用者線程自己來執(zhí)行這個(gè)任務(wù)。這提供了一個(gè)簡(jiǎn)單的反饋機(jī)制會(huì)拖慢調(diào)用者從而降低新任務(wù)的提交速度是一種平緩的削峰方式。DiscardOldestPolicy丟棄隊(duì)列里最老的一個(gè)任務(wù)然后嘗試執(zhí)行當(dāng)前任務(wù)。這可能會(huì)丟失重要任務(wù)。DiscardPolicy默默丟棄無法處理的任務(wù)不拋異常。風(fēng)險(xiǎn)最大。生產(chǎn)環(huán)境建議通常使用AbortPolicy并結(jié)合業(yè)務(wù)層的降級(jí)、熔斷機(jī)制。或者自定義拒絕策略比如將拒絕的任務(wù)持久化到磁盤、發(fā)到死信隊(duì)列待系統(tǒng)恢復(fù)后重試或者至少記錄詳細(xì)的日志和告警。3.2 線程池工作流程與調(diào)優(yōu)實(shí)戰(zhàn)線程池處理任務(wù)遵循一個(gè)固定的流程理解這個(gè)流程是調(diào)優(yōu)的基礎(chǔ)提交一個(gè)新任務(wù)。如果當(dāng)前運(yùn)行的線程數(shù) corePoolSize則立即創(chuàng)建新線程執(zhí)行該任務(wù)即使有空閑核心線程此策略也可能創(chuàng)建新線程取決于具體實(shí)現(xiàn)但通常優(yōu)先使用空閑線程。如果運(yùn)行的線程數(shù) corePoolSize則嘗試將任務(wù)放入workQueue。如果隊(duì)列已滿且運(yùn)行的線程數(shù) maximumPoolSize則創(chuàng)建新線程非核心執(zhí)行任務(wù)。如果隊(duì)列已滿且運(yùn)行的線程數(shù)已達(dá)maximumPoolSize則觸發(fā)RejectedExecutionHandler。調(diào)優(yōu)實(shí)戰(zhàn)案例一個(gè)訂單處理服務(wù)假設(shè)我們有一個(gè)訂單處理服務(wù)任務(wù)是CPU密集型計(jì)算優(yōu)惠、庫存校驗(yàn)等。機(jī)器配置4核CPU。初步設(shè)置corePoolSize 4 1 5,maximumPoolSize 10。隊(duì)列選擇使用ArrayBlockingQueue容量設(shè)為100。拒絕策略自定義將拒絕的訂單ID記錄到Redis或發(fā)到Kafka后續(xù)補(bǔ)償。上線后通過監(jiān)控如Micrometer Prometheus發(fā)現(xiàn)線程數(shù)長(zhǎng)期在5-6個(gè)隊(duì)列很少堆積。說明核心線程數(shù)設(shè)置基本合理。在促銷期間監(jiān)控到有任務(wù)被拒絕。分析日志發(fā)現(xiàn)拒絕發(fā)生在流量尖峰持續(xù)約2秒。優(yōu)化此時(shí)不應(yīng)盲目調(diào)大線程數(shù)CPU密集型任務(wù)線程太多反而因頻繁上下文切換導(dǎo)致性能下降。我們采取的措施是優(yōu)化任務(wù)本身分析被拒絕的任務(wù)看是否有計(jì)算邏輯可以優(yōu)化縮短單個(gè)任務(wù)處理時(shí)間。擴(kuò)容隊(duì)列將隊(duì)列容量從100調(diào)整為200以應(yīng)對(duì)更短暫的尖峰。但要注意隊(duì)列容量太大會(huì)增加任務(wù)延遲。完善降級(jí)在自定義拒絕策略中除了記錄立即給用戶返回“系統(tǒng)繁忙請(qǐng)稍后再試”的友好提示并觸發(fā)異步補(bǔ)償流程。3.3 線程池的關(guān)閉與監(jiān)控正確關(guān)閉shutdown()和shutdownNow()。shutdown()溫和關(guān)閉。不再接受新任務(wù)但會(huì)執(zhí)行完已提交的任務(wù)和隊(duì)列中的任務(wù)。shutdownNow()暴力關(guān)閉。嘗試中斷所有正在執(zhí)行的任務(wù)不再處理隊(duì)列中的任務(wù)返回尚未開始執(zhí)行的任務(wù)列表。最佳實(shí)踐通常先調(diào)用shutdown()然后awaitTermination等待一段時(shí)間如果超時(shí)仍有任務(wù)未完成再調(diào)用shutdownNow()。executor.shutdown(); try { if (!executor.awaitTermination(60, TimeUnit.SECONDS)) { executor.shutdownNow(); if (!executor.awaitTermination(60, TimeUnit.SECONDS)) { System.err.println(線程池未能正常終止); } } } catch (InterruptedException ie) { executor.shutdownNow(); Thread.currentThread().interrupt(); // 保留中斷狀態(tài) }監(jiān)控指標(biāo)生產(chǎn)環(huán)境必須監(jiān)控線程池。活動(dòng)線程數(shù)反映當(dāng)前忙碌程度。隊(duì)列大小反映任務(wù)積壓情況。已完成任務(wù)數(shù)反映吞吐量。拒絕任務(wù)數(shù)這是最重要的告警指標(biāo)之一說明系統(tǒng)已過載。 可以利用ThreadPoolExecutor自帶的方法getActiveCount(),getQueue().size()等來暴露這些指標(biāo)到你的監(jiān)控系統(tǒng)。4.CompletableFuture異步編程的“瑞士軍刀”在Java 8之前處理異步任務(wù)主要靠Future但它獲取結(jié)果的方式是阻塞的get()方法且難以描述任務(wù)間的依賴關(guān)系如“任務(wù)A和B都完成后再執(zhí)行C”。CompletableFuture的出現(xiàn)讓Java擁有了強(qiáng)大的函數(shù)式異步編程能力。4.1 核心概念創(chuàng)建與簡(jiǎn)單轉(zhuǎn)換創(chuàng)建異步任務(wù)// 1. 使用默認(rèn)的 ForkJoinPool.commonPool() 執(zhí)行 CompletableFutureString future1 CompletableFuture.supplyAsync(() - { // 模擬耗時(shí)計(jì)算 try { Thread.sleep(1000); } catch (InterruptedException e) { } return Result from supplyAsync; }); // 2. 使用自定義線程池生產(chǎn)環(huán)境推薦 ExecutorService customExecutor Executors.newFixedThreadPool(5); CompletableFutureString future2 CompletableFuture.supplyAsync(() - { return Result with custom executor; }, customExecutor);注意supplyAsync用于有返回值的任務(wù)runAsync用于無返回值的任務(wù)。強(qiáng)烈建議為CPU密集型或重要的IO任務(wù)指定自定義線程池避免所有CompletableFuture共享同一個(gè)公共池導(dǎo)致相互影響。處理計(jì)算結(jié)果thenApply,thenAccept,thenRun這三個(gè)方法是鏈?zhǔn)秸{(diào)用的基礎(chǔ)分別代表轉(zhuǎn)換、消費(fèi)和執(zhí)行。thenApply(FunctionT, U)接收上一個(gè)任務(wù)的結(jié)果進(jìn)行轉(zhuǎn)換返回新的CompletableFutureU。thenAccept(ConsumerT)接收結(jié)果進(jìn)行消費(fèi)如打印、保存不返回新結(jié)果。thenRun(Runnable)不關(guān)心上一個(gè)任務(wù)的結(jié)果只是在前一個(gè)階段完成后執(zhí)行一個(gè)動(dòng)作。CompletableFuture.supplyAsync(() - Hello) .thenApply(s - s World) // 轉(zhuǎn)換得到 Hello World .thenApply(String::toUpperCase) // 轉(zhuǎn)換得到 HELLO WORLD .thenAccept(System.out::println) // 消費(fèi)打印結(jié)果 .thenRun(() - System.out.println(All done.)); // 執(zhí)行打印完成信息關(guān)鍵點(diǎn)這些方法都有對(duì)應(yīng)的異步版本thenApplyAsync等它們會(huì)將后續(xù)的任務(wù)提交到線程池中執(zhí)行而不是由完成上一個(gè)任務(wù)的線程直接執(zhí)行。這可以避免某個(gè)耗時(shí)任務(wù)阻塞整個(gè)鏈。4.2 組合任務(wù)描述復(fù)雜的依賴關(guān)系這是CompletableFuture最強(qiáng)大的地方。thenCompose扁平化依賴類似flatMap用于串聯(lián)兩個(gè)有依賴關(guān)系的異步任務(wù)第二個(gè)任務(wù)需要第一個(gè)任務(wù)的結(jié)果。// 模擬先根據(jù)用戶ID查詢用戶信息再根據(jù)用戶信息中的地址ID查詢地址 CompletableFutureUser userFuture getUserAsync(userId); CompletableFutureAddress addressFuture userFuture.thenCompose(user - { return getAddressAsync(user.getAddressId()); // 此操作返回一個(gè)新的CompletableFutureAddress }); // addressFuture 最終完成時(shí)得到的是地址對(duì)象而不是嵌套的Future。thenCombine合并兩個(gè)獨(dú)立任務(wù)的結(jié)果兩個(gè)異步任務(wù)并行執(zhí)行當(dāng)它們都完成后對(duì)它們的結(jié)果進(jìn)行合并處理。CompletableFutureInteger futureA getPriceAsync(itemA); CompletableFutureDouble futureB getDiscountAsync(user123); CompletableFutureDouble totalPriceFuture futureA.thenCombine(futureB, (price, discount) - { return price * discount; // 合并計(jì)算最終價(jià)格 });allOf/anyOf等待多個(gè)任務(wù)allOf(CompletableFuture?... cfs)返回一個(gè)新的Future當(dāng)所有給定的Future都完成時(shí)它才完成。它沒有結(jié)果值常用于等待一批并行任務(wù)全部結(jié)束。CompletableFutureVoid allFutures CompletableFuture.allOf(future1, future2, future3); allFutures.thenRun(() - { // 所有任務(wù)都完成了可以執(zhí)行后續(xù)操作比如匯總結(jié)果 // 注意要獲取各個(gè)future的結(jié)果仍需調(diào)用 future1.join() 等 });anyOf(CompletableFuture?... cfs)返回一個(gè)新的Future當(dāng)任意一個(gè)給定的Future完成時(shí)它就完成其結(jié)果與最先完成的那個(gè)Future相同。可用于實(shí)現(xiàn)“競(jìng)速”或超時(shí)備用。4.3 異常處理與超時(shí)控制異常處理exceptionally和handleexceptionally(FunctionThrowable, T)相當(dāng)于catch當(dāng)鏈中之前的階段出現(xiàn)異常時(shí)提供一個(gè)新的返回值。CompletableFuture.supplyAsync(() - { if (new Random().nextBoolean()) { throw new RuntimeException(Oops!); } return Success; }).exceptionally(ex - { System.err.println(Error: ex.getMessage()); return Default Value; // 提供降級(jí)值 }).thenAccept(System.out::println);handle(BiFunctionT, Throwable, U)無論成功還是異常都會(huì)執(zhí)行它同時(shí)接收結(jié)果和異常可以統(tǒng)一處理。.handle((result, ex) - { if (ex ! null) { return Handled Error: ex.getMessage(); } return Result: result; })超時(shí)控制Java 9 Java 9為CompletableFuture增加了orTimeout和completeOnTimeout方法使得超時(shí)處理變得異常簡(jiǎn)單。CompletableFutureString future CompletableFuture.supplyAsync(() - { try { Thread.sleep(2000); } catch (InterruptedException e) { } return Result; }) .orTimeout(1, TimeUnit.SECONDS) // 設(shè)置1秒超時(shí)超時(shí)后拋出 TimeoutException .exceptionally(ex - Fallback due to timeout: ex.getClass().getSimpleName());對(duì)于Java 8需要通過completeOnTimeout或與ScheduledExecutorService配合來實(shí)現(xiàn)超時(shí)。實(shí)戰(zhàn)心得CompletableFuture的鏈?zhǔn)秸{(diào)用雖然優(yōu)雅但過長(zhǎng)的鏈和復(fù)雜的組合會(huì)降低代碼可讀性。在復(fù)雜的業(yè)務(wù)流中可以考慮將其拆分成多個(gè)有命名意義的方法。另外要小心回調(diào)地獄雖然CompletableFuture比純回調(diào)好但嵌套過深依然難以維護(hù)。對(duì)于非常復(fù)雜的異步流程可以考慮使用響應(yīng)式編程庫如Project Reactor。5. 鎖的進(jìn)階ReentrantLock與AQS窺探synchronized是JVM內(nèi)置的鎖簡(jiǎn)單易用。而ReentrantLock作為JUC提供的顯式鎖提供了更靈活、更強(qiáng)大的功能。5.1ReentrantLock的核心優(yōu)勢(shì)可中斷的鎖獲取lockInterruptibly()方法允許在等待鎖的過程中響應(yīng)中斷這對(duì)于實(shí)現(xiàn)可取消的任務(wù)非常重要。嘗試非阻塞獲取鎖tryLock()方法嘗試獲取鎖如果鎖被占用它不會(huì)阻塞而是立即返回false。可以用于避免死鎖或?qū)崿F(xiàn)某些特定邏輯。公平鎖與非公平鎖ReentrantLock的構(gòu)造器可以指定是否創(chuàng)建公平鎖。公平鎖保證等待時(shí)間最長(zhǎng)的線程優(yōu)先獲取鎖避免了“饑餓”但會(huì)帶來更大的性能開銷因?yàn)樾枰S護(hù)一個(gè)有序隊(duì)列。非公平鎖是默認(rèn)的也是性能更高的選擇在大多數(shù)高并發(fā)場(chǎng)景下推薦使用。綁定多個(gè)條件一個(gè)ReentrantLock可以創(chuàng)建多個(gè)Condition對(duì)象實(shí)現(xiàn)更精細(xì)的線程間通信。synchronized只能有一個(gè)等待集wait/notifyAll。5.2 抽象隊(duì)列同步器AQS淺析ReentrantLock、Semaphore、CountDownLatch等許多JUC同步工具其底層都依賴于一個(gè)共同的框架——AbstractQueuedSynchronizer (AQS)。理解AQS有助于我們看清這些工具的本質(zhì)。AQS的核心思想是它維護(hù)了一個(gè)volatile int state同步狀態(tài)和一個(gè)FIFO線程等待隊(duì)列CLH隊(duì)列的變體。對(duì)于不同的同步器state的含義不同。對(duì)于ReentrantLockstate表示鎖被重入的次數(shù)對(duì)于Semaphorestate表示剩余的許可數(shù)量對(duì)于CountDownLatchstate表示倒計(jì)數(shù)的初始值。同步器需要重寫AQS的tryAcquire、tryRelease等方法來定義如何獲取和釋放狀態(tài)。當(dāng)線程嘗試獲取狀態(tài)失敗時(shí)AQS會(huì)將線程封裝成節(jié)點(diǎn)加入隊(duì)列并可能阻塞該線程。當(dāng)狀態(tài)釋放時(shí)AQS會(huì)負(fù)責(zé)喚醒隊(duì)列中的后繼線程。以ReentrantLock的非公平鎖實(shí)現(xiàn)為例lock()方法首先會(huì)直接嘗試用CAS將state從0改為1快速路徑如果成功就將當(dāng)前線程設(shè)為獨(dú)占所有者。這體現(xiàn)了“非公平”性新來的線程可能比隊(duì)列中等待的線程先拿到鎖。如果快速路徑失敗則調(diào)用AQS的acquire方法最終會(huì)調(diào)用子類重寫的tryAcquire再次嘗試如果還失敗就將線程加入隊(duì)列并可能掛起。學(xué)習(xí)建議對(duì)于大多數(shù)應(yīng)用開發(fā)者無需深究AQS的每一個(gè)細(xì)節(jié)。但了解其基本原理能讓你在遇到復(fù)雜的同步問題時(shí)知道該從哪個(gè)方向去查閱源碼和資料也能更好地理解那些基于AQS構(gòu)建的工具的行為。這是從“會(huì)用”到“懂原理”的關(guān)鍵一步。6. 實(shí)戰(zhàn)避坑與性能調(diào)優(yōu)經(jīng)驗(yàn)錄理論最終要服務(wù)于實(shí)踐。下面是我在多年高并發(fā)項(xiàng)目開發(fā)中總結(jié)的一些常見“坑”和調(diào)優(yōu)經(jīng)驗(yàn)。6.1 線程安全與可見性那些容易忽略的細(xì)節(jié)“單例模式”的雙重檢查鎖DCL陷阱與正確寫法老生常談但依然有人寫錯(cuò)。錯(cuò)誤的DCL在于instance new Singleton()這行代碼不是原子的它可能發(fā)生指令重排導(dǎo)致其他線程拿到一個(gè)未初始化完全的對(duì)象。// 錯(cuò)誤示例在舊版本Java內(nèi)存模型下有問題 public class Singleton { private static Singleton instance; public static Singleton getInstance() { if (instance null) { // 第一次檢查 synchronized (Singleton.class) { if (instance null) { // 第二次檢查 instance new Singleton(); // 問題在此 } } } return instance; } }正確寫法方法一最簡(jiǎn)潔利用類加載機(jī)制推薦。public class Singleton { private static class Holder { private static final Singleton INSTANCE new Singleton(); } public static Singleton getInstance() { return Holder.INSTANCE; } }方法二使用volatile關(guān)鍵字JDK5。public class Singleton { private static volatile Singleton instance; public static Singleton getInstance() { if (instance null) { synchronized (Singleton.class) { if (instance null) { instance new Singleton(); } } } return instance; } }volatile不能保證復(fù)合操作的原子性如前所述volatile只保證可見性和禁止指令重排。count這種“讀-改-寫”操作必須使用synchronized或原子類。線程局部變量ThreadLocal的內(nèi)存泄漏ThreadLocal的經(jīng)典內(nèi)存泄漏場(chǎng)景使用線程池時(shí)線程是復(fù)用的。如果ThreadLocal變量用完后沒有調(diào)用remove()清理那么該線程的ThreadLocalMap中會(huì)一直保留對(duì)value的強(qiáng)引用Entry的key是弱引用但value是強(qiáng)引用導(dǎo)致value對(duì)象無法被回收。最佳實(shí)踐在使用完ThreadLocal后務(wù)必在finally塊中調(diào)用threadLocal.remove()。6.2 死鎖與活鎖的診斷與預(yù)防死鎖四個(gè)必要條件互斥、持有并等待、不可剝奪、循環(huán)等待。預(yù)防死鎖的核心是打破循環(huán)等待。一個(gè)實(shí)用的方法是定義鎖的獲取順序所有線程都按相同的全局順序申請(qǐng)鎖。// 定義鎖的順序 private static final Object lock1 new Object(); private static final Object lock2 new Object(); public void method1() { synchronized (lock1) { // 先獲取lock1 synchronized (lock2) { // 再獲取lock2 // do something } } } public void method2() { synchronized (lock1) { // 同樣先獲取lock1即使它只需要lock2 synchronized (lock2) { // do something else } } }活鎖線程沒有阻塞但在不斷重試某個(gè)總是失敗的操作比如兩個(gè)線程互相謙讓資源導(dǎo)致誰都無法進(jìn)行。解決方案是引入隨機(jī)退避時(shí)間。6.3 性能調(diào)優(yōu)監(jiān)控指標(biāo)線上高并發(fā)系統(tǒng)必須監(jiān)控以下與線程相關(guān)的指標(biāo)線程狀態(tài)通過jstack或Arthas等工具定期查看線程狀態(tài)分布。大量的BLOCKED或WAITING線程可能是鎖競(jìng)爭(zhēng)激烈或IO等待的征兆。鎖競(jìng)爭(zhēng)使用jstack查看線程等待的鎖或使用JMX、ReentrantLock的getQueueLength()等方法監(jiān)控等待特定鎖的線程數(shù)。CPU使用率與上下文切換過高的上下文切換vmstat中的cs列意味著線程過多或鎖競(jìng)爭(zhēng)激烈。結(jié)合pidstat或top -H查看具體進(jìn)程和線程的CPU使用情況。GC情況不當(dāng)?shù)牟l(fā)對(duì)象創(chuàng)建如在循環(huán)中new大量臨時(shí)對(duì)象會(huì)導(dǎo)致Young GC頻繁甚至引發(fā)Full GC。監(jiān)控GC頻率和耗時(shí)。6.4 虛擬線程Java 21的展望Java 21引入的虛擬線程Virtual Threads是并發(fā)編程的一次重大革新。它由JVM管理非常輕量初始內(nèi)存約幾百字節(jié)可以創(chuàng)建數(shù)百萬個(gè)而不會(huì)導(dǎo)致系統(tǒng)資源耗盡。其目標(biāo)是用簡(jiǎn)單的同步阻塞代碼風(fēng)格獲得異步非阻塞的高性能。 對(duì)于傳統(tǒng)的、大量時(shí)間花在等待IO如數(shù)據(jù)庫查詢、網(wǎng)絡(luò)調(diào)用上的業(yè)務(wù)代碼可以幾乎不做修改只需將ExecutorService換成Executors.newVirtualThreadPerTaskExecutor()就能獲得巨大的吞吐量提升因?yàn)樗鼘⒆枞腛S線程釋放出來去執(zhí)行其他虛擬線程的任務(wù)。當(dāng)前建議如果你的項(xiàng)目已使用Java 21并且是IO密集型應(yīng)用強(qiáng)烈建議開始評(píng)估和測(cè)試虛擬線程。但對(duì)于CPU密集型任務(wù)或依賴現(xiàn)有復(fù)雜線程池調(diào)優(yōu)邏輯的應(yīng)用遷移需謹(jǐn)慎。虛擬線程是未來但理解好今天的平臺(tái)線程Thread和JUC是擁抱這個(gè)未來的堅(jiān)實(shí)基礎(chǔ)。高并發(fā)編程是一個(gè)既需要深厚理論支撐又需要大量實(shí)戰(zhàn)經(jīng)驗(yàn)積累的領(lǐng)域。JUC提供了一套強(qiáng)大的工業(yè)級(jí)工具但工具本身不會(huì)寫出好代碼。真正的關(guān)鍵在于你是否理解每個(gè)工具背后的設(shè)計(jì)意圖、適用場(chǎng)景和潛在陷阱并能在復(fù)雜的業(yè)務(wù)邏輯中做出恰當(dāng)?shù)倪x擇和組合。希望這篇來自一線的萬字心得能成為你工具箱里一件稱手的兵器助你在高并發(fā)的戰(zhàn)場(chǎng)上更加游刃有余。記住沒有銀彈持續(xù)學(xué)習(xí)、謹(jǐn)慎實(shí)踐、重視監(jiān)控才是應(yīng)對(duì)并發(fā)挑戰(zhàn)的不二法門。