通過本文檔你將學(xué)習(xí)到

• 線程初始化的幾種方式亡鼠。
• 線程的相關(guān)概念
• 線程池怎么玩
• 線程池的常見面試題

1 線程回顧知識

1.1 線程初始化的幾種方式

1)铁坎、繼承 Thread
2)凿掂、實現(xiàn) Runnable 接口
3)遗菠、實現(xiàn) Callable 接口 + FutureTask (可以拿到返回結(jié)果痴荐，可以處理異常)
4)列肢、線程池
常見線程方式一

public static class Thread01 extends Thread {
    @Override
    public void run() {
        System.out.println("當(dāng)前線程：" + Thread.currentThread().getId());
        int i = 10 / 2;
        System.out.println("運行結(jié)果：" + i);
    }
}
Thread thread = new Thread01();
 thread.start();

常見線程方式二

public static class Runable01 implements Runnable {
    @Override
    public void run() {
        System.out.println("當(dāng)前線程：" + Thread.currentThread().getId());
        int i = 10 / 2;
        System.out.println("運行結(jié)果：" + i);
    }
}

 Runable01 runable01 = new Runable01();
 new Thread(runable01).start();

說到上面兩種恰画，大家非常非常的熟悉，就不多扯淡了,接下來瓷马，將引出第三種Callable拴还。

Thread thread = new Thread01();
 thread.start();

相信這個大家再熟悉不過了，剛?cè)腴T的時候欧聘，就是這么學(xué)習(xí)的片林。接下來我們就打開API

Thread

一般情況下我們都是使用 Thread(Runnable target, String name)吧

廢話不多說 要是有一個這構(gòu)造方法 Thread(Callable target, String name) 那不是直接搞起么
那么說明孝偎，我們一拍腦袋半秒就能想到的問題访敌，這sum Oracle公司的天才們都想不到，
他們就是蠢貨一個衣盾，馬上寫信給他們說寺旺，讓我來拯救JAVA代碼
那么接下來你有兩種做法，馬上發(fā)郵件告訴sum Oracle公司的天才們
1 我發(fā)現(xiàn)了BUG,你們沒有擴(kuò)展一個方法 Thread(Callable target, String name)
2 自動誕生的時候天才們就想到了擴(kuò)展方法势决，根本不用多余的加個構(gòu)造方法阻塑，最后發(fā)現(xiàn)小丑竟然是自己
你覺得是哪一個呢？

回歸正題：本來我只能接受Runnable 果复，現(xiàn)在Callable 說陈莽，我也想勾搭下Thread，那么你準(zhǔn)備怎么搞虽抄？
你想想搞定Thread走搁，但是我不認(rèn)識，那我就先搞定它周人的人迈窟，間接不就搞定了么私植？
你想想我們寫代碼的時候參數(shù)都是寫的list 然后傳遞list的子類ArrayList對吧。更底層的代碼直接傳遞的Collection车酣，有時候大神寫代碼你看著更無語曲稼，直接傳遞一個接口，這個接口還沒有任何方法湖员，繼承后也不用實現(xiàn)贫悄，這種混個身份，到時候可以透傳的娘摔，如果你這是無語了窄坦，。你要說這人是傻逼凳寺，到最后發(fā)現(xiàn)小丑竟是我自己對吧嫡丙？因為我們能力還達(dá)不到人家的境界，無法理解读第。等你能力提高后，你才會發(fā)現(xiàn)人家為什么這么寫底層代碼拥刻？哦明白了怜瞒，看著好自然，對，就該這么搞吴汪』菡看著代碼像大自然一樣自然

FutureTask 就是中間人了，幫你撮合
Thread(Callable target, String name) 沒有 這個構(gòu)造漾橙，那么中間人過來
Thread(中間人 FutureTask , String name) 
new  Thread(futureTask , '"我的名字叫線程110號").star() 

在此之前也得介紹下：Future
這個是一個接口杆融，獲取異步任務(wù)的結(jié)果，取消任務(wù)的執(zhí)行霜运，判斷任務(wù)是否被取消脾歇，判斷任務(wù)執(zhí)行是否完畢。
主線程讓子線程去執(zhí)行其它事情淘捡。Future接口的實現(xiàn)類FutureTask 藕各。

image.png

JDK.5以后常見方式三：

public static class Callable01 implements Callable <Integer> {
    @Override
    public Integer call() throws Exception {
        System.out.println("當(dāng)前線程：" + Thread.currentThread().getId());
        int i = 10 / 2;
        System.out.println("運行結(jié)果：" + i);
        return I;
    }
}
我啟動了1000個線程去跑對賬任務(wù)，現(xiàn)在編號999線程出錯了焦除，你給我返回錯誤激况，我想看下為什么？

  FutureTask<Integer> futureTask = new FutureTask<>(new Callable01());
 new Thread(futureTask).start();
 System.out.println(futureTask.get());

方式4 直接提交給線程池即可

public static ExecutorService executor = Executors.newFixedThreadPool(10);
// service.execute(new Runable01());
// Future<Integer> submit = service.submit(new Callable01());
// submit.get();

futureTask優(yōu)勢明顯呀惹骂，你看搭配線程池用的飛起瑞妇，那么它又缺點嗎逻住？
當(dāng)然有呀，人無完人浙踢，代碼也有缺點
1、futureTask.get()阻塞 也可以傳遞x秒 等待幾秒
2蹈丸、那我用isDone()方法成黄。

image.png

這樣也不行呀，浪費cpu字段逻杖，你一直while true 要有個回調(diào)通知那多好呀奋岁。
一般情況下我們都是通過輪詢的方式獲取結(jié)果，進(jìn)來不要阻塞荸百。

那有沒有什么方式解決上面的缺點呢闻伶？
答案是肯定的，有的够话，我們不用這個類蓝翰，我們有一個更好的選擇 CompletableFuture 這個比Future還能干，F(xiàn)uture能干的CompletableFuture也可以女嘲，CompletableFuture還能干Future不能干的畜份。爽不爽？

CompletableFuture基本介紹欣尼。
阻塞的方式和異步編程的設(shè)計理念相違背爆雹，而輪詢的方式會消耗無畏的CPU資源停蕉。因此，JDK8設(shè)計出CompletableFuture

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {

在Java 8中钙态， Complet able Future提供了非常強(qiáng)大的Future的擴(kuò)展功能慧起， 可以幫助我們簡化異步編程的復(fù)雜性， 并且提供了函數(shù)式編程的能力册倒， 可以通過回調(diào)的方式處理計算結(jié)果蚓挤， 也提供了轉(zhuǎn)換和組合Complet able Future的方法。
它可能代表一個明確完成的Future驻子， 也有可能代表一個完成階段(Completion Stage) 灿意， 它支持在計算完成以后觸發(fā)一些函數(shù)或執(zhí)行某些動作。

?核心的四個靜態(tài)方法

public static CompletableFuture<Void> runAsync(Runnable runnable
runAsync+線程池
public static CompletableFuture<Void> runAsync(Runnable runnable,
                                                   Executor executor)
supplyAsync 有返回值
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) 
 supplyAsync+線程池
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
                                                       Executor executor)

CompletableFuture優(yōu)點總結(jié)

異步任務(wù)結(jié)束時拴孤，會自動回調(diào)某個對象的方法脾歧；

主線程設(shè)置好后，不再關(guān)心異步任務(wù)的執(zhí)行演熟，異步任務(wù)之間可以順序執(zhí)行

異步任務(wù)出錯時鞭执，會自動回調(diào)某個對象的方法

1.2 為什么要使用線程池

1、2芒粹、 不能得到返回值
3 可以得到返回值但是他們都不能控制資源
實際玩的過程中我們都用線程池 第四方式兄纺。

1.3 線程池的介紹

線程池做的工作主要是控制運行的線程的數(shù)量,處理過程中將任務(wù)加入隊列,然后在線程創(chuàng)建后啟動這些任務(wù),如果線程超過了最大數(shù)量,超出的數(shù)量的線程排隊等候,等其他線程執(zhí)行完畢,再從隊列中取出任務(wù)來執(zhí)行.

他的主要特點為:線程復(fù)用:控制最大并發(fā)數(shù):管理線程.

第一:降低資源消耗.通過重復(fù)利用自己創(chuàng)建的線程降低線程創(chuàng)建和銷毀造成的消耗.
第二: 提高響應(yīng)速度.當(dāng)任務(wù)到達(dá)時,任務(wù)可以不需要等到線程和粗昂就愛你就能立即執(zhí)行.
第三: 提高線程的可管理性.線程是稀缺資源,如果無限的創(chuàng)建,不僅會消耗資源,還會較低系統(tǒng)的穩(wěn)定性,使用線程池可以進(jìn)行統(tǒng)一分配,調(diào)優(yōu)和監(jiān)控.

1.4 線程池如何使用?

Java中的線程池是通過Executor框架實現(xiàn)的,該框架中用到了Executor,Executors,ExecutorService,ThreadPoolExecutor這幾個類.

image.png

1.5 面試重點 說下常見的線程池唄

Executors.newFixedThreadPool(int)
執(zhí)行一個長期的任務(wù),性能好很多

  public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

主要特點如下:
1.創(chuàng)建一個定長線程池,可控制線程的最大并發(fā)數(shù),超出的線程會在隊列中等待.
2.newFixedThreadPool創(chuàng)建的線程池corePoolSize和MaxmumPoolSize是 相等的,它使用的的LinkedBlockingQueue

Executors.newSingleThreadExecutor()
一個任務(wù)一個線程執(zhí)行的任務(wù)場景

 public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

主要特點如下:
1.創(chuàng)建一個單線程化的線程池,它只會用唯一的工作線程來執(zhí)行任務(wù),保證所有任務(wù)都按照指定順序執(zhí)行.
2.newSingleThreadExecutor將corePoolSize和MaxmumPoolSize都設(shè)置為1,它使用的的LinkedBlockingQueue

Executors.newCachedThreadPool()
適用:執(zhí)行很多短期異步的小程序或者負(fù)載較輕的服務(wù)器

   public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

主要特點如下:
1.創(chuàng)建一個可緩存線程池,如果線程池長度超過處理需要,可靈活回收空閑線程,若無可回收,則創(chuàng)建新線程.
2.newCachedThreadPool將corePoolSize設(shè)置為0MaxmumPoolSize設(shè)置為Integer.MAX_VALUE,它使用的是SynchronousQUeue,也就是說來了任務(wù)就創(chuàng)建線程運行,如果線程空閑超過60秒,就銷毀線程

1.6 面試重點 說下常見的線程池的七大參數(shù)唄

 public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) 

從源碼中可以看出，線程池的構(gòu)造函數(shù)有7個參數(shù)化漆，分別是corePoolSize估脆、maximumPoolSize、keepAliveTime座云、unit疙赠、workQueue、threadFactory朦拖、handler圃阳。下面會對這7個參數(shù)一一解釋。

一璧帝、corePoolSize 線程池核心線程大小

線程池中會維護(hù)一個最小的線程數(shù)量捍岳，即使這些線程處理空閑狀態(tài)，他們也不會被銷毀睬隶，除非設(shè)置了allowCoreThreadTimeOut锣夹。這里的最小線程數(shù)量即是corePoolSize。

1.在創(chuàng)建了線程池后,當(dāng)有請求任務(wù)來之后,就會安排池中的線程去執(zhí)行請求任務(wù),近視理解為今日當(dāng)值線程
2.當(dāng)線程池中的線程數(shù)目達(dá)到corePoolSize后,就會把到達(dá)的任務(wù)放入到緩存隊列當(dāng)中.

二苏潜、maximumPoolSize 線程池最大線程數(shù)量

一個任務(wù)被提交到線程池以后银萍，首先會找有沒有空閑存活線程，如果有則直接將任務(wù)交給這個空閑線程來執(zhí)行恤左，如果沒有則會緩存到工作隊列（后面會介紹）中贴唇，如果工作隊列滿了贰锁，才會創(chuàng)建一個新線程，然后從工作隊列的頭部取出一個任務(wù)交由新線程來處理滤蝠，而將剛提交的任務(wù)放入工作隊列尾部。線程池不會無限制的去創(chuàng)建新線程授嘀，它會有一個最大線程數(shù)量的限制物咳，這個數(shù)量即由maximunPoolSize指定。

三蹄皱、keepAliveTime 空閑線程存活時間

一個線程如果處于空閑狀態(tài)览闰，并且當(dāng)前的線程數(shù)量大于corePoolSize，那么在指定時間后巷折，這個空閑線程會被銷毀压鉴，這里的指定時間由keepAliveTime來設(shè)定

默認(rèn)情況下:
只有當(dāng)線程池中的線程數(shù)大于corePoolSize時keepAliveTime才會起作用,知道線程中的線程數(shù)不大于corepoolSIze,
四、unit 空閑線程存活時間單位

keepAliveTime的計量單位

五锻拘、workQueue 工作隊列

新任務(wù)被提交后油吭，會先進(jìn)入到此工作隊列中，任務(wù)調(diào)度時再從隊列中取出任務(wù)署拟。jdk中提供了四種工作隊列：

①ArrayBlockingQueue

基于數(shù)組的有界阻塞隊列婉宰，按FIFO排序。新任務(wù)進(jìn)來后推穷，會放到該隊列的隊尾心包，有界的數(shù)組可以防止資源耗盡問題。當(dāng)線程池中線程數(shù)量達(dá)到corePoolSize后馒铃，再有新任務(wù)進(jìn)來蟹腾，則會將任務(wù)放入該隊列的隊尾，等待被調(diào)度区宇。如果隊列已經(jīng)是滿的娃殖，則創(chuàng)建一個新線程，如果線程數(shù)量已經(jīng)達(dá)到maxPoolSize萧锉，則會執(zhí)行拒絕策略珊随。

②LinkedBlockingQuene

基于鏈表的無界阻塞隊列（其實最大容量為Interger.MAX），按照FIFO排序柿隙。由于該隊列的近似無界性叶洞，當(dāng)線程池中線程數(shù)量達(dá)到corePoolSize后，再有新任務(wù)進(jìn)來禀崖，會一直存入該隊列衩辟，而不會去創(chuàng)建新線程直到maxPoolSize，因此使用該工作隊列時波附，參數(shù)maxPoolSize其實是不起作用的艺晴。

③SynchronousQuene

一個不緩存任務(wù)的阻塞隊列昼钻，生產(chǎn)者放入一個任務(wù)必須等到消費者取出這個任務(wù)。也就是說新任務(wù)進(jìn)來時封寞，不會緩存然评，而是直接被調(diào)度執(zhí)行該任務(wù)，如果沒有可用線程狈究，則創(chuàng)建新線程碗淌，如果線程數(shù)量達(dá)到maxPoolSize，則執(zhí)行拒絕策略抖锥。

④PriorityBlockingQueue

具有優(yōu)先級的無界阻塞隊列亿眠，優(yōu)先級通過參數(shù)Comparator實現(xiàn)。

六磅废、threadFactory 線程工廠

創(chuàng)建一個新線程時使用的工廠纳像，可以用來設(shè)定線程名、是否為daemon線程等等

七拯勉、handler 拒絕策略

當(dāng)工作隊列中的任務(wù)已到達(dá)最大限制竟趾，并且線程池中的線程數(shù)量也達(dá)到最大限制，這時如果有新任務(wù)提交進(jìn)來谜喊，該如何處理呢潭兽。這里的拒絕策略，就是解決這個問題的斗遏，jdk中提供了4中拒絕策略：
AbortPolicy(默認(rèn)):直接拋出RejectedException異常阻止系統(tǒng)正常運行
CallerRunPolicy:"調(diào)用者運行"一種調(diào)節(jié)機(jī)制,該策略既不會拋棄任務(wù),也不會拋出異常,而是將任務(wù)分給調(diào)用線程來執(zhí)行
DiscardOldestPolicy:拋棄隊列中等待最久的任務(wù),然后把當(dāng)前任務(wù)加入隊列中嘗試再次提交
DiscardPolicy:直接丟棄任務(wù),不予任何處理也不拋出異常.如果允許任務(wù)丟失,這是最好的拒絕策略
以上內(nèi)置策略均實現(xiàn)了RejectExecutionHandler接口

一句圖總結(jié)中心思想:面試必問

image.png

1.7 坑你沒商量

你在工作中單一的/固定數(shù)的/可變你的三種創(chuàng)建線程池的方法,你用哪個多?超級大坑

答案是一個都不用,我們生產(chǎn)上只能使用自定義的
Executors中JDK給你提供了為什么不用?

【強(qiáng)制】線程資源必須通過線程池提供山卦，不允許在應(yīng)用中自行顯式創(chuàng)建線程。
說明：使用線程池的好處是減少在創(chuàng)建和銷毀線程上所消耗的時間以及系統(tǒng)資源的開銷诵次，解決資源不足的問題账蓉。如果不使用線程池，有可能造成系統(tǒng)創(chuàng)建大量同類線程而導(dǎo)致消耗完內(nèi)存或者“過度切換”的問題逾一。

【強(qiáng)制】線程池不允許使用Executors去創(chuàng)建铸本，而是通過ThreadPoolExecutor的方式，這樣的處理方式讓寫的同學(xué)更加明確線程池的運行規(guī)則遵堵，規(guī)避資源耗盡的風(fēng)險箱玷。說明：Executors返回的線程池對象的弊端如下：
1）FixedThreadPool和SingleThreadPool:允許的請求隊列長度為Integer.MAX_VALUE，可能會堆積大量的請求陌宿，從而導(dǎo)致OOM锡足。
2）CachedThreadPool和ScheduledThreadPool:允許的創(chuàng)建線程數(shù)量為Integer.MAX_VALUE，可能會創(chuàng)建大量的線程壳坪，從而導(dǎo)致OOM舶得。

1.7 你在工作中是如何創(chuàng)建線程池的,是否自定義過線程池使用

public class MyThreadPoolDemo {
    public static void main(String[] args) {
        ExecutorService threadPool = new ThreadPoolExecutor(
                2,
                5,
                1L,
                TimeUnit.SECONDS,
                new LinkedBlockingDeque<Runnable>(3),
                Executors.defaultThreadFactory(),
                //默認(rèn)拋出異常
                //new ThreadPoolExecutor.AbortPolicy()
                //回退調(diào)用者
                //new ThreadPoolExecutor.CallerRunsPolicy()
                //處理不來的不處理
                //new ThreadPoolExecutor.DiscardOldestPolicy()
                new ThreadPoolExecutor.DiscardPolicy()
        );
        //模擬10個用戶來辦理業(yè)務(wù) 沒有用戶就是來自外部的請求線程.
        try {
            for (int i = 1; i <= 10; i++) {
                threadPool.execute(() -> {
                    System.out.println(Thread.currentThread().getName() + "\t 辦理業(yè)務(wù)");
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool.shutdown();
        }
        //threadPoolInit();
    }

    private static void threadPoolInit() {
        /**
         * 一池5個處理線程
         */
        //ExecutorService threadPool= Executors.newFixedThreadPool(5);
        /**
         * 一池一線程
         */
        //ExecutorService threadPool= Executors.newSingleThreadExecutor();
        /**
         * 一池N線程
         */
        ExecutorService threadPool = Executors.newCachedThreadPool();
        //模擬10個用戶來辦理業(yè)務(wù) 沒有用戶就是來自外部的請求線程.
        try {
            for (int i = 1; i <= 20; i++) {
                threadPool.execute(() -> {
                    System.out.println(Thread.currentThread().getName() + "\t 辦理業(yè)務(wù)");
                });
                try {
                    TimeUnit.MICROSECONDS.sleep(200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool.shutdown();
        }
    }
}

此處看下RMQ的源碼以及自己項目中的使用情況

1.7 合理配置線程池你是如何考慮的?

image.png

CPU 密集型
核心數(shù)+1。

但是為什么要加一呢爽蝴？《Java并發(fā)編程實戰(zhàn)》一書中給出的原因是：即使當(dāng)計算（CPU）密集型的線程偶爾由于頁缺失故障或者其他原因而暫停時沐批，這個“額外”的線程也能確保 CPU 的時鐘周期不會被浪費纫骑。
看不懂是不是？沒關(guān)系我也看不懂九孩。反正把它理解為一個備份的線程就行了先馆。

IO密集型
cpu*2

image.png

既是IO密集型也是CPU密集型
合理分配即可

以上都是面試答案，但是
線程池使用面臨的核心的問題在于：線程池的參數(shù)并不好配置躺彬。
一方面線程池的運行機(jī)制不是很好理解磨隘，配置合理需要強(qiáng)依賴開發(fā)人員的個人經(jīng)驗和知識；
另一方面顾患，線程池執(zhí)行的情況和任務(wù)類型相關(guān)性較大，IO密集型和CPU密集型的任務(wù)運行起來的情況差異非常大个唧。
這導(dǎo)致業(yè)界并沒有一些成熟的經(jīng)驗策略幫助開發(fā)人員參考江解。
https://blog.csdn.net/qq_41893274/article/details/112424767

1.8數(shù)據(jù)賞析

RMQ的BrokerController類等等

       //5、啟動定時調(diào)度徙歼，每10秒鐘掃描所有Broker犁河，檢查存活狀態(tài)
 private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
        "NSScheduledThread"));

        this.scheduledExecutorService.scheduleAtFixedRate(() -> NamesrvController.this.routeInfoManager.scanNotActiveBroker(), 5, 10, TimeUnit.SECONDS);
//不是說不能用嗎？你還用魄梯？為什么桨螺？



//周期性向nameserv發(fā)心跳，如果有變化則同步broker信息

 private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl(
        "BrokerControllerScheduledThread"));


   this.scheduledExecutorService.scheduleAtFixedRate(() -> {
                try {
                    BrokerController.this.getBrokerStats().record();
                } catch (Throwable e) {
                    log.error("schedule record error.", e);
                }
            }, initialDelay, period, TimeUnit.MILLISECONDS);

Broker向所有的nameServer注冊自己的戶口本信息酿秸。

   private BrokerFixedThreadPoolExecutor brokerOuterExecutor = new BrokerFixedThreadPoolExecutor(4, 10, 1, TimeUnit.MINUTES,
        new ArrayBlockingQueue<Runnable>(32), new ThreadFactoryImpl("brokerOutApi_thread_", true));


  final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
            for (final String namesrvAddr : nameServerAddressList) {
                brokerOuterExecutor.execute(() -> {
                    try {
                        RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);
                        if (result != null) {
                            registerBrokerResultList.add(result);
                        }

                        log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);
                    } catch (Exception e) {
                        log.warn("registerBroker Exception, {}", namesrvAddr, e);
                    } finally {
                        countDownLatch.countDown();
                    }
                });
            }

            try {
                countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) {
            }