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