本節(jié)要點(diǎn)

• 了解一些線程安全的案例
• 學(xué)習(xí)線程安全的設(shè)計(jì)模型
• 掌握單例模式,阻塞隊(duì)列,生產(chǎn)在消費(fèi)者模型

單例模式

我們知道多線程編程,因?yàn)榫€程的隨機(jī)調(diào)度會(huì)出現(xiàn)很多線程安全問題! 而我們的java有些大佬針對(duì)一些多線程安全問題的應(yīng)用場(chǎng)景,設(shè)計(jì)了一些對(duì)應(yīng)的解決方法和案例,就是解決這些問題的一些套路,被稱為設(shè)計(jì)模式,供我們學(xué)習(xí)和使用!

單例模式是校招最常考的一個(gè)設(shè)計(jì)模式之一!!!

什么是單例模式呢?

單例模式能保證某個(gè)類在程序中只存在唯一一份實(shí)例, 而不會(huì)創(chuàng)建出多個(gè)實(shí)例.
這一點(diǎn)在很多場(chǎng)景上都需要. 比如 JDBC 中的 DataSource 實(shí)例就只需要一個(gè)

單例模式的具體實(shí)現(xiàn)方法又分為餓漢和懶漢兩種!
而這里所說的餓和懶并不是貶義詞!
餓漢指的是在創(chuàng)建一個(gè)類的時(shí)候就將實(shí)例創(chuàng)建好!比較急!
懶漢指的是在需要用到實(shí)例的時(shí)候再去創(chuàng)建實(shí)例!比較懶!

餓漢模式

餓漢模式聯(lián)系實(shí)際生活中例子:
就是一個(gè)人性子比較急,也許一件事情的期限還有好久,而他卻把事情早早干完!

因?yàn)槲覀儐卫Ｊ街荒苡幸粋€(gè)實(shí)例
那如何去保證一個(gè)實(shí)例呢?
我們會(huì)馬上想到類中用static修飾的類屬性,它只有一份!保證了單例模式的基本條件!

顯然生活中這樣的人很優(yōu)秀,但是我們的計(jì)算機(jī)如果這樣卻不太好!
因?yàn)?code>cpu和內(nèi)存的空間有限,如果還不需要用到該實(shí)例,卻創(chuàng)建了實(shí)例,那不就增加了內(nèi)存開銷,顯然不科學(xué).但事實(shí)問題也不大!

class Singleton{
    //餓漢模式, static 創(chuàng)建類時(shí),就創(chuàng)建好了類屬性的實(shí)例!
    //private 這里的instance實(shí)例只有一份!!!
    private static Singleton instance = new Singleton();
    //私有的構(gòu)造方法!保證該實(shí)例不能再創(chuàng)建
    private Singleton(){
    }
    //提供一個(gè)方法,外界可以獲取到該實(shí)例!
    public static Singleton getInstance() {
        return instance;
    }
}

我們可以看到這里餓漢模式,當(dāng)多個(gè)線程并發(fā)時(shí),并沒有出現(xiàn)線程不安全問題,因?yàn)檫@里的設(shè)計(jì)模式只是針對(duì)了讀操作!!! 而單例模式的更改操作,需要看懶漢模式!

懶漢模式

聯(lián)系實(shí)際中的例子就是.就是這個(gè)人比較拖延,有些事情不得不做的時(shí)候,他才會(huì)去做完!

//懶漢模式(線程不安全版本)
class Singleton1{
    //懶漢模式, static 創(chuàng)建類時(shí),并沒有創(chuàng)建實(shí)例!
    //private 保證這里的instance實(shí)例只有一份!!!
    private static Singleton1 instance = null;
    //私有的構(gòu)造方法!保證該實(shí)例不能再創(chuàng)建
    private Singleton1(){
    }
    //提供一個(gè)方法,外界可以獲取到該實(shí)例!
    public static Singleton1 getInstance() {
        if(instance==null){//需要時(shí)再創(chuàng)建實(shí)例!
            instance = new Singleton1();
        }
        return instance;
    }
}

我們分析一下上述代碼,該模式,對(duì)singleton進(jìn)行了修改,而我們知道多線程的修改可能會(huì)出現(xiàn)線程不安全問題!
當(dāng)我們多個(gè)線程同時(shí)對(duì)該變量進(jìn)行訪問時(shí)!

我們將該代碼的情況分成兩種,一種是初始化前要進(jìn)行讀寫操作,初始化后只需要進(jìn)行讀操作!

• instance未初始化化前
  多個(gè)線程同時(shí)進(jìn)入getInstance方法!那就會(huì)創(chuàng)建很多次instance實(shí)例!
  聯(lián)系之前的變量更改內(nèi)存和cpu的操作:
  
  顯然很多線程進(jìn)行了無效操作!!!也會(huì)觸發(fā)內(nèi)存不可見問題!!!
• instance初始化后,進(jìn)行的讀操作,就像上面的餓漢模式一樣,并沒有線程安全問題!

我們下面進(jìn)行多次優(yōu)化

//優(yōu)化1
class Singleton2{
    //懶漢模式, static 創(chuàng)建類時(shí),并沒有創(chuàng)建實(shí)例!
    //private 保證這里的instance實(shí)例只有一份!!!
    private static Singleton2 instance = null;
    //私有的構(gòu)造方法!保證該實(shí)例不能再創(chuàng)建
    private Singleton2(){
    }
    //提供一個(gè)方法,外界可以獲取到該實(shí)例!
    public static Singleton2 getInstance() {
        synchronized (Singleton.class){ //對(duì)讀寫操作進(jìn)行加鎖!
            if(instance==null){//需要時(shí)再創(chuàng)建實(shí)例!
                instance = new Singleton2();
            }
            return instance;
        }
    }
}

我們將Singleton類對(duì)象加鎖后,顯然避免了剛剛的一些線程安全問題!但是出現(xiàn)了新的問題!

• instance初始化前
  在初始化前,我們很好的將讀寫操作進(jìn)行了原子封裝,并不會(huì)造成線程不安全問題!
• instance初始化后
  然而初始化后的每次讀操作卻并不好,當(dāng)我們多個(gè)線程進(jìn)行多操作時(shí),很多線程就會(huì)造成線程阻塞,代碼的運(yùn)行效率極具下降!

我們?nèi)绾伪ＷC,線程安全的情況下又保證讀操作不會(huì)進(jìn)行加鎖,鎖競(jìng)爭(zhēng)呢?

我們可以間代碼的兩種情況分別處理!

//優(yōu)化二
class Singleton2{
    //懶漢模式, static 創(chuàng)建類時(shí),并沒有創(chuàng)建實(shí)例!
    //private 保證這里的instance實(shí)例只有一份!!!
    private static Singleton2 instance = null;
    //私有的構(gòu)造方法!保證該實(shí)例不能再創(chuàng)建
    private Singleton2(){
    }
    //提供一個(gè)方法,外界可以獲取到該實(shí)例!
    public static Singleton2 getInstance() {
        if(instance==null){//如果未初始化就進(jìn)行加鎖操作!
            synchronized (Singleton.class){ //對(duì)讀寫操作進(jìn)行加鎖!
                if(instance==null){//需要時(shí)再創(chuàng)建實(shí)例!
                    instance = new Singleton2();
                }
            }
        }
        //已經(jīng)初始化后直接讀!!!
        return instance;
    }
}

我們看到這里可能會(huì)有疑惑,咋為啥要套兩個(gè)if啊,把里面的if刪除不行嗎!!!
我們來看刪除后的效果:

//刪除里層if
class Singleton2{
    //懶漢模式, static 創(chuàng)建類時(shí),并沒有創(chuàng)建實(shí)例!
    //private 保證這里的instance實(shí)例只有一份!!!
    private static Singleton2 instance = null;
    //私有的構(gòu)造方法!保證該實(shí)例不能再創(chuàng)建
    private Singleton2(){
    }
    //提供一個(gè)方法,外界可以獲取到該實(shí)例!
    public static Singleton2 getInstance() {
        if(instance==null){//如果未初始化就進(jìn)行加鎖操作!
            synchronized (Singleton.class){ //對(duì)讀寫操作進(jìn)行加鎖!
                    instance = new Singleton2();
            }
        }
        //已經(jīng)初始化后直接讀!!!
        return instance;
    }
}

在刪除里層的if后:
我們發(fā)現(xiàn)當(dāng)有多個(gè)線程進(jìn)行了第一個(gè)if判斷后,進(jìn)入的線程中有一個(gè)線程鎖競(jìng)爭(zhēng)拿到了鎖!而其他線程就在這阻塞等待,直到該鎖釋放后,又有線程拿到了該鎖,而這樣也就多次創(chuàng)建了instance實(shí)例,顯然不可!!!

所以這里的兩個(gè)if都有自己的作用缺一不可!
第一個(gè)if:
判斷是否要進(jìn)行加鎖初始化
第二個(gè)if:
判斷該線程實(shí)例是否已經(jīng)創(chuàng)建!

//最終優(yōu)化版
class Singleton2{
    //懶漢模式, static 創(chuàng)建類時(shí),并沒有創(chuàng)建實(shí)例!
    //private 保證這里的instance實(shí)例只有一份!!!
    //volatile 保證內(nèi)存可見!!!避免編譯器優(yōu)化!!!
    private static volatile Singleton2 instance = null;
    //私有的構(gòu)造方法!保證該實(shí)例不能再創(chuàng)建
    private Singleton2(){
    }
    //提供一個(gè)方法,外界可以獲取到該實(shí)例!
    public static Singleton2 getInstance() {
        if(instance==null){//如果未初始化就進(jìn)行加鎖操作!
            synchronized (Singleton.class){ //對(duì)讀寫操作進(jìn)行加鎖!
	            if(instance==null){
	                instance = new Singleton2();
	            }
            }
        }
        //已經(jīng)初始化后直接讀!!!
        return instance;
    }
}

而我們又發(fā)現(xiàn)了一個(gè)問題,我們的編譯器是會(huì)對(duì)代碼進(jìn)行優(yōu)化操作的!如果很多線程對(duì)第一個(gè)if進(jìn)行判斷,那cpu老是在內(nèi)存中拿instance的值,就很慢,編譯器就不開心了,它就優(yōu)化直接將該值存在寄存器中,而此操作是否危險(xiǎn),如果有一個(gè)線程將該實(shí)例創(chuàng)建!那就會(huì)導(dǎo)致線程安全問題! 而volatile關(guān)鍵字保證了instanse內(nèi)存可見性!!!

總結(jié)懶漢模式

• 雙if 外層保證未初始化前加鎖,創(chuàng)建實(shí)例. 里層if保證實(shí)例創(chuàng)建唯一一次
• synchronized加鎖,保證讀寫原子性
• volatile保證內(nèi)存可見性,避免編譯器優(yōu)化

阻塞隊(duì)列

什么是阻塞隊(duì)列?
顧名思義是隊(duì)列的一種!
也符合先進(jìn)先出的特點(diǎn)!
阻塞隊(duì)列特點(diǎn):

當(dāng)隊(duì)列為空時(shí),讀操作阻塞
當(dāng)隊(duì)列為滿時(shí),寫操作阻塞

阻塞隊(duì)列一般用在多線程中!并且有很多的應(yīng)用場(chǎng)景!
最典型的一個(gè)應(yīng)用場(chǎng)景就是生產(chǎn)者消費(fèi)者模型

生產(chǎn)者消費(fèi)者模型

我們知道生產(chǎn)者和消費(fèi)者有著供需關(guān)系!
而開發(fā)中很多場(chǎng)景都會(huì)有這樣的供需關(guān)系!
比如有兩個(gè)服務(wù)器A和B
A是入口服務(wù)器直接接受用戶的網(wǎng)絡(luò)請(qǐng)求
B應(yīng)用服務(wù)器對(duì)A進(jìn)行數(shù)據(jù)提供

在通常情況下如果一個(gè)網(wǎng)站的訪問量不大,那么A和B服務(wù)器都能正常使用!
而我們知道,很多網(wǎng)站當(dāng)很多用戶進(jìn)行同時(shí)訪問時(shí)就可能掛!
我們知道,A入口服務(wù)器和B引用服務(wù)器此時(shí)耦合度較高!
當(dāng)一個(gè)掛了,那么另一個(gè)服務(wù)器也會(huì)出現(xiàn)問題!

而當(dāng)我們使用生產(chǎn)者消費(fèi)者模型就很好的解決了上述高度耦合問題!我們?cè)谒麄冎虚g加入一個(gè)阻塞隊(duì)列即可!

當(dāng)增加就緒隊(duì)列后,我們就不用擔(dān)心A和B的耦合!
并且A和B進(jìn)行更改都不會(huì)影響到對(duì)方! 甚至將改變服務(wù)器,對(duì)方也無法察覺!
而阻塞隊(duì)列還保證了,服務(wù)器的訪問速度,不管用戶量多大! 這些數(shù)據(jù)都會(huì)先傳入阻塞隊(duì)列,而阻塞隊(duì)列如果滿,或者空,都會(huì)線程阻塞! 也就不存在服務(wù)器爆了的問題!!!
也就是起到了削峰填谷的作用!不管訪問量一時(shí)間多大!就緒隊(duì)列都可以保證服務(wù)器的速度!

標(biāo)準(zhǔn)庫(kù)中的就緒隊(duì)列

我們java中提供了一組就緒隊(duì)列供我們使用!

BlockingQueue

BlockingQueue 是一個(gè)接口. 真正實(shí)現(xiàn)的類是 LinkedBlockingQueue.
put 方法用于阻塞式的入隊(duì)列,
take 用于阻塞式的出隊(duì)列.
BlockingQueue 也有 offer, poll, peek 等方法, 但是這些方法不帶有阻塞特性.

//生產(chǎn)著消費(fèi)者模型
public class Test2 {
    public static void main(String[] args) throws InterruptedException {
        //創(chuàng)建一個(gè)阻塞隊(duì)列
        BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<Integer>();
        Thread customer = new Thread(() -> {//消費(fèi)者
            while (true) {
                try {
                    int value = blockingQueue.take();
                    System.out.println("消費(fèi)元素: " + value);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "消費(fèi)者");
        customer.start();
        Thread producer = new Thread(() -> {//生產(chǎn)者
            Random random = new Random();
            while (true) {
                try {
                    int num = random.nextInt(1000);
                    System.out.println("生產(chǎn)元素: " + num);
                    blockingQueue.put(num);
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "生產(chǎn)者");
        producer.start();
        customer.join();
        producer.join();
    }
}

阻塞隊(duì)列實(shí)現(xiàn)

雖然java標(biāo)準(zhǔn)庫(kù)中提供了阻塞隊(duì)列,但是我們想自己實(shí)現(xiàn)一個(gè)阻塞隊(duì)列!

我們就用循環(huán)隊(duì)列實(shí)現(xiàn)吧,使用數(shù)組!

//循環(huán)隊(duì)列
class MyblockingQueue{
    //阻塞隊(duì)列
    private int[] data = new int[100];
    //隊(duì)頭
    private int start = 0;
    //隊(duì)尾
    private int tail = 0;
    //元素個(gè)數(shù), 用于判斷隊(duì)列滿
    private int size = 0;
    public void put(int x){
        //入隊(duì)操作
        if(size==data.length){
            //隊(duì)列滿
            return;
        }
        data[tail] = x;
        tail++;//入隊(duì)
        if(tail==data.length){
            //判斷是否需要循環(huán)回
            tail=0;
        }
        size++; //入隊(duì)成功加1
    }
    public Integer take(){
        //出隊(duì)并且獲取隊(duì)頭元素
        if(tail==start){
            //隊(duì)列為空!
            return null;
        }
        int ret = data[start]; //獲取隊(duì)頭元素
        start++; //出隊(duì)
        if(start==data.length){
            //判斷是否要循環(huán)回來
            start = 0;
        }
       // start = start % data.length;//不建議可讀性不搞,效率也低
        size--;//元素個(gè)數(shù)減一
        return ret;
    }
}

我們已經(jīng)創(chuàng)建好了一個(gè)循環(huán)隊(duì)列,目前達(dá)不到阻塞的效果!
而且當(dāng)多線程并發(fā)時(shí)有很多線程不安全問題!
而我們知道想要阻塞,那不得加鎖,不然哪來的阻塞!

//阻塞隊(duì)列
class MyblockingQueue{
    //阻塞隊(duì)列
    private int[] data = new int[100];
    //隊(duì)頭
    private int start = 0;
    //隊(duì)尾
    private int tail = 0;
    //元素個(gè)數(shù), 用于判斷隊(duì)列滿
    private int size = 0;
    //鎖對(duì)象
    Object locker = new Object();

    public void put(int x){
       synchronized (locker){//對(duì)該操作加鎖
           //入隊(duì)操作
           if(size==data.length){
               //隊(duì)列滿 阻塞等待!!!直到put操作后notify才會(huì)繼續(xù)執(zhí)行
               try {
                   locker.wait();
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }
           data[tail] = x;
           tail++;//入隊(duì)
           if(tail==data.length){
               //判斷是否需要循環(huán)回
               tail=0;
           }
           size++; //入隊(duì)成功加1
           //入隊(duì)成功后通知take 如果take阻塞
           locker.notify();//這個(gè)操作線程阻塞并沒有副作用!
       }
    }
    public Integer take(){
        //出隊(duì)并且獲取隊(duì)頭元素
        synchronized (locker){
            if(size==0){
                //隊(duì)列為空!阻塞等待 知道隊(duì)列有元素put就會(huì)繼續(xù)執(zhí)行該線程
                try {
                    locker.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            int ret = data[start]; //獲取隊(duì)頭元素
            start++; //出隊(duì)
            if(start==data.length){
                //判斷是否要循環(huán)回來
                start = 0;
            }
            // start = start % data.length;//不建議可讀性不搞,效率也低
            size--;//元素個(gè)數(shù)減一
            locker.notify();//通知 put 如果put阻塞!
            return ret;
        }
    }
}

//測(cè)試代碼
public class Test3 {
    public static void main(String[] args) {
            MyblockingQueue queue = new MyblockingQueue();
            Thread customer = new Thread(()->{
                int i = 0;
                while (true){
                    System.out.println("消費(fèi)了"+queue.take());
                }
            });

                    Thread producer = new Thread(()->{
                        Random random = new Random();
                        while (true){
                            int x = random.nextInt(100);
                            System.out.println("生產(chǎn)了"+x);
                            queue.put(x);
                            try {
                                Thread.sleep(100);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }
                    });
                    customer.start();
                    producer.start();
    }
}

可以看到通過wait和notify的配和,我就實(shí)現(xiàn)了阻塞隊(duì)列!!!

定時(shí)器

定時(shí)器是什么

定時(shí)器也是軟件開發(fā)中的一個(gè)重要組件. 類似于一個(gè) “鬧鐘”. 達(dá)到一個(gè)設(shè)定的時(shí)間之后, 就執(zhí)行某個(gè)指定好的代碼.

也就是說定時(shí)器有像join和sleep等待功能,不過他們是基于系統(tǒng)內(nèi)部的定時(shí)器,
而我們要學(xué)習(xí)的是在java給我們提供的定時(shí)器包裝類,用于到了指定時(shí)間就執(zhí)行代碼!
并且定時(shí)器在我們?nèi)粘ｉ_發(fā)中十分常用!

java給我們提供了專門一個(gè)定時(shí)器的封裝類Timer在java.util包下!

Timer定時(shí)器

Timer類下有一個(gè)schedule方法,用于安排指定的任務(wù)和執(zhí)行時(shí)間!
也就達(dá)到了定時(shí)的效果,如果時(shí)間到了,就會(huì)執(zhí)行task!

• schedule 包含兩個(gè)參數(shù).
• 第一個(gè)參數(shù)指定即將要執(zhí)行的任務(wù)代碼,
• 第二個(gè)參數(shù)指定多長(zhǎng)時(shí)間之后執(zhí)行 (單位為毫秒).

//實(shí)例
import java.util.Timer;
import java.util.TimerTask;
public class Demo1 {
    public static void main(String[] args) {
        //在java.util.Timer包下
        Timer timer = new Timer();
        //timer.schedule()方法傳入需要執(zhí)行的任務(wù)和定時(shí)時(shí)間
        //Timer內(nèi)部有專門的線程負(fù)責(zé)任務(wù)的注冊(cè),所以不需要start
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                System.out.println("hello Timer!");
            }
        },3000);
        //main線程
        System.out.println("hello main!");
    }
}

我們可以看到我們只需要?jiǎng)?chuàng)建一個(gè)Timer對(duì)象,然后調(diào)用schedule返回,傳入你要執(zhí)行的任務(wù),和定時(shí)時(shí)間便可完成!

定時(shí)器實(shí)現(xiàn)

我們居然知道java中定時(shí)器的使用,那如何自己實(shí)現(xiàn)一個(gè)定時(shí)器呢!

我們可以通過Timer中的源碼,然后進(jìn)行操作!

Timer內(nèi)部需要什么東西呢!

我們想想Timer的功能!
可以定時(shí)執(zhí)行任務(wù)!(線程)
可以知道任務(wù)啥時(shí)候執(zhí)行(時(shí)間)
可以將多個(gè)任務(wù)組織起來對(duì)比時(shí)間執(zhí)行

• 描述任務(wù)
  也就是schedule方法中傳入的TimerTake
  創(chuàng)建一個(gè)專門表示定時(shí)器中的任務(wù)

class MyTask{
    //任務(wù)具體要干啥
    private Runnable runnable;
    //任務(wù)執(zhí)行時(shí)間,時(shí)間戳
    private long time;
    ///delay是一個(gè)時(shí)間間隔
    public MyTask(Runnable runnable,long delay){
            this.runnable = runnable;
            time = System.currentTimeMillis()+delay;
    }
    public void run(){ //描述任務(wù)!
        runnable.run();
    }
}

• 組織任務(wù)
  組織任務(wù)就是將上述的任務(wù)組織起來!
  我們知道我們的任務(wù)需要在多線程的環(huán)境下執(zhí)行,所以就需要有線程安全,阻塞功能的數(shù)據(jù)結(jié)構(gòu)!并且我們的任務(wù)到了時(shí)間就需要執(zhí)行,也就是需要時(shí)刻對(duì)任務(wù)排序!
  所以我們采用PriorityBlockingQueue優(yōu)先級(jí)隊(duì)列!阻塞!
  
  但是這里我們使用了優(yōu)先級(jí)隊(duì)列,我們需要指定比較規(guī)則,就是讓MyTask實(shí)現(xiàn)Comparable接口,重寫 compareTo方法,指定升序排序,就是小根堆!
  

• 執(zhí)行時(shí)間到了的任務(wù)
  我們可以創(chuàng)建一個(gè)線程,執(zhí)行時(shí)間到了的任務(wù)!

//執(zhí)行時(shí)間到了的任務(wù)!
    public MyTimer(){
        Thread thread = new Thread(()->{
           while (true){
               try {
                   MyTask task = queue.take();//獲取到隊(duì)首任務(wù)
                   //比較時(shí)間是否到了
                   //獲取當(dāng)前時(shí)間戳
                   long curTime = System.currentTimeMillis();
                   if(curTime<task.getTime()){//當(dāng)前時(shí)間戳和該任務(wù)需要執(zhí)行的時(shí)間比較
                       //還未到達(dá)執(zhí)行時(shí)間
                       queue.put(task); //將任務(wù)放回
                   }else{//時(shí)間到了,執(zhí)行任務(wù)
                       task.run();
                   }
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }
        });
        thread.start();//啟動(dòng)線程!
    }

//定時(shí)器完整代碼
import java.util.concurrent.PriorityBlockingQueue;
class MyTask implements Comparable<MyTask>{
    //任務(wù)具體要干啥
    private Runnable runnable;

    public long getTime() {
        return time;
    }

    //任務(wù)執(zhí)行時(shí)間,時(shí)間戳
    private long time;
    ///delay是一個(gè)時(shí)間間隔
    public MyTask(Runnable runnable,long delay){
            this.runnable = runnable;
            time = System.currentTimeMillis()+delay;
    }
    public void run(){ //描述任務(wù)!
        runnable.run();
    }
    @Override
    public int compareTo(MyTask o) {
        return (int)(this.time - o.time);
    }
}
public class MyTimer{
    //定時(shí)器內(nèi)部需要存放多個(gè)任務(wù)
    private PriorityBlockingQueue<MyTask> queue = new PriorityBlockingQueue<>();
    public void schedule(Runnable runnable,long delay){
        MyTask task = new MyTask(runnable,delay);//接收一個(gè)任務(wù)!
        queue.put(task);//將任務(wù)組織起來
    }
   //執(zhí)行時(shí)間到了的任務(wù)!
    public MyTimer(){
        Thread thread = new Thread(()->{
           while (true){
               try {
                   MyTask task = queue.take();//獲取到隊(duì)首任務(wù)
                   //比較時(shí)間是否到了
                   //獲取當(dāng)前時(shí)間戳
                   long curTime = System.currentTimeMillis();
                   if(curTime<task.getTime()){//當(dāng)前時(shí)間戳和該任務(wù)需要執(zhí)行的時(shí)間比較
                       //還未到達(dá)執(zhí)行時(shí)間
                       queue.put(task); //將任務(wù)放回
                   }else{//時(shí)間到了,執(zhí)行任務(wù)
                       task.run();
                   }
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }
        });
        thread.start();//啟動(dòng)線程!
    }
}

//測(cè)試
public static void main(String[] args) {
        MyTimer myTimer = new MyTimer();
        myTimer.schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println("hello Timer");
            }
        }, 3000);
        System.out.println("hello main");
    }

我們?cè)賮頇z查一下下面代碼存在的問題!

 //執(zhí)行時(shí)間到了的任務(wù)!
    public MyTimer(){
        Thread thread = new Thread(()->{
           while (true){
               try {
                   MyTask task = queue.take();//獲取到隊(duì)首任務(wù)
                   //比較時(shí)間是否到了
                   //獲取當(dāng)前時(shí)間戳
                   long curTime = System.currentTimeMillis();
                   if(curTime<task.getTime()){//當(dāng)前時(shí)間戳和該任務(wù)需要執(zhí)行的時(shí)間比較
                       //還未到達(dá)執(zhí)行時(shí)間
                       queue.put(task); //將任務(wù)放回
                   }else{//時(shí)間到了,執(zhí)行任務(wù)
                       task.run();
                   }
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }
        });
        thread.start();//啟動(dòng)線程!
    }
 

我們上述代碼還存在一定缺陷就是執(zhí)行線程到了的代碼,我們的while循環(huán)一直在處于忙等狀態(tài)!
就好比生活中:
你9點(diǎn)要去做核酸,然后你過一會(huì)就看時(shí)間,一會(huì)就看時(shí)間,感覺就有啥大病一樣!
所以我們可以定一個(gè)鬧鐘,到了時(shí)間就去,沒到時(shí)間可以干其他的事情!
此處的線程也是如此!我們這里也可以使用wait阻塞! 然后到了時(shí)間就喚醒,就解決了忙等問題!
我們的wait可以傳入指定的時(shí)間,到了該時(shí)間就喚醒!!!

我們?cè)偎伎剂硪粋€(gè)問題!

如果又加入了新的任務(wù)呢?
我們此時(shí)也需要喚醒一下線程,讓線程重新拿到隊(duì)首元素!

//最終定時(shí)器代碼!!!!
import java.util.concurrent.PriorityBlockingQueue;
class MyTask implements Comparable<MyTask>{
    //任務(wù)具體要干啥
    private Runnable runnable;

    public long getTime() {
        return time;
    }
    //任務(wù)執(zhí)行時(shí)間,時(shí)間戳
    private long time;
    ///delay是一個(gè)時(shí)間間隔
    public MyTask(Runnable runnable,long delay){
            this.runnable = runnable;
            time = System.currentTimeMillis()+delay;
    }
    public void run(){ //描述任務(wù)!
        runnable.run();
    }
    @Override
    public int compareTo(MyTask o) {
        return (int)(this.time - o.time);
    }
}
public class MyTimer{
    //定時(shí)器內(nèi)部需要存放多個(gè)任務(wù)
    Object locker = new Object();//鎖對(duì)象
    private PriorityBlockingQueue<MyTask> queue = new PriorityBlockingQueue<>();
    public void schedule(Runnable runnable,long delay){
        MyTask task = new MyTask(runnable,delay);//接收一個(gè)任務(wù)!
        queue.put(task);//將任務(wù)組織起來

        //每次拿到新的任務(wù)就需要喚醒線程,重新得到新的隊(duì)首元素!
        synchronized (locker){
            locker.notify();
        }
    }
   //執(zhí)行時(shí)間到了的任務(wù)!
    public MyTimer(){
        Thread thread = new Thread(()->{
           while (true){
               try {
                   MyTask task = queue.take();//獲取到隊(duì)首任務(wù)
                   //比較時(shí)間是否到了
                   //獲取當(dāng)前時(shí)間戳
                   long curTime = System.currentTimeMillis();
                   if(curTime<task.getTime()){//當(dāng)前時(shí)間戳和該任務(wù)需要執(zhí)行的時(shí)間比較
                       //還未到達(dá)執(zhí)行時(shí)間
                       queue.put(task); //將任務(wù)放回
                       //阻塞到該時(shí)間喚醒!
                       synchronized (locker){
                           locker.wait(task.getTime()-curTime);
                       }
                   }else{//時(shí)間到了,執(zhí)行任務(wù)
                       task.run();
                   }
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }
        });
        thread.start();//啟動(dòng)線程!
    }
}

總結(jié):

• 描述一個(gè)任務(wù) runnable + time
• 使用優(yōu)先級(jí)隊(duì)列組織任務(wù)PriorityBlockingQueue
• 實(shí)現(xiàn)schedule方法來注冊(cè)任務(wù)到隊(duì)列
• 創(chuàng)建掃描線程,獲取隊(duì)首元素,判斷是否執(zhí)行
• 注意這里的忙等問題

//最后梳理一遍
import java.util.concurrent.PriorityBlockingQueue;
/**
 * Created with IntelliJ IDEA.
 * Description:定時(shí)器
 * User: hold on
 * Date: 2022-04-09
 * Time: 16:07
 */
//1.描述任務(wù)
class Task implements Comparable<Task>{
    //任務(wù)
    private Runnable runnable;
    //執(zhí)行時(shí)間
    private long time;
    public Task(Runnable runnable,long delay){
        this.runnable = runnable;//傳入任務(wù)
        //獲取任務(wù)需要執(zhí)行的時(shí)間戳
        time = System.currentTimeMillis() + delay;
    }
    @Override
    public int compareTo(Task o) {//指定比較方法!
        return (int) (this.time-o.time);
    }

    public long getTime() {//傳出任務(wù)時(shí)間
        return time;
    }
    public void run(){
        runnable.run();
    }
}
//組織任務(wù)
class MyTimer1{
    private Object locker = new Object();//鎖對(duì)象
    //用于組織任務(wù)
    private PriorityBlockingQueue<Task> queue = new PriorityBlockingQueue<>();
    public void schedule(Runnable runnable,long delay){
        Task task = new Task(runnable,delay);
        queue.put(task);//傳入隊(duì)列中
        synchronized (locker){
            locker.notify();//喚醒線程
        }
    }
    public MyTimer1(){
        //掃描線程獲取隊(duì)首元素,判斷執(zhí)行
        Thread thread = new Thread(()->{
           while (true){
               //獲取當(dāng)前時(shí)間戳
               long curTimer = System.currentTimeMillis();
               try {
                   Task task = queue.take();//隊(duì)首元素出隊(duì)
                   if(curTimer<task.getTime()){
                       //還未到達(dá)執(zhí)行時(shí)間,返回隊(duì)首元素
                       queue.put(task);
                       synchronized (locker){
                           //阻塞等待
                           locker.wait(task.getTime()-curTimer);
                       }
                   }else {
                       //執(zhí)行任務(wù)
                       task.run();
                   }
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }
        });
        thread.start();//啟動(dòng)線程
    }
}
public class Demo2 {
    public static void main(String[] args) {
        MyTimer1 myTimer1 = new MyTimer1();
        myTimer1.schedule(new Runnable() {
            @Override
            public void run() {
                System.out.println("hello Timer1");
            }
        },1000);

        System.out.println("hello main");
    }
}

線程池

我們之前學(xué)過常量池!這里的線程池也大同小異!
我們通過創(chuàng)建很多個(gè)線程放在一塊空間不進(jìn)行銷毀,等到需要的時(shí)候就啟動(dòng)線程!避免了創(chuàng)建銷毀的時(shí)間開銷! 提高開發(fā)效率!
我們之前不是說一個(gè)線程創(chuàng)建并不會(huì)劃分很多時(shí)間嗎! 但是我們的多線程編程,有時(shí)候需要使用到很多很多線程,如果要進(jìn)行創(chuàng)建,效率就不高,而線程池或者協(xié)程(我們后面會(huì)介紹)就避免了創(chuàng)建銷毀線程! 但我們需要用到線程時(shí),自己從線程池中給出就好!

我們創(chuàng)建線程的本質(zhì)還是要通過內(nèi)核態(tài)(就是我們的操作系統(tǒng))進(jìn)行創(chuàng)建,然而內(nèi)核態(tài)創(chuàng)建的時(shí)間,我們程序員無法掌控,而通過線程池,我們就可以避免了內(nèi)核態(tài)的操作,直接在用戶態(tài),進(jìn)行線程的調(diào)用,也就是應(yīng)用程序?qū)?

使用線程池大大提高了我們的開發(fā)效率!

我們來學(xué)習(xí)一下java中給我們提供的線程池類,然后自己實(shí)現(xiàn)一個(gè)線程池!

ThreadPoolExecutor 線程池

這個(gè)類在java.util.concurrent 并發(fā)編程包下,我們用到的很多關(guān)于并發(fā)編程的類都在!

可以看到這個(gè)線程池有4個(gè)構(gòu)造方法!

我們了解一下參數(shù)最多的那個(gè)方法!

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

創(chuàng)建一個(gè)新的 ThreadPoolExecutor與給定的初始參數(shù)。

參數(shù)
    corePoolSize - 即使空閑時(shí)仍保留在池中的線程數(shù)，除非設(shè)置 allowCoreThreadTimeOut 
    maximumPoolSize - 池中允許的最大線程數(shù) 
    keepAliveTime - 當(dāng)線程數(shù)大于內(nèi)核時(shí)，這是多余的空閑線程在終止前等待新任務(wù)的最大時(shí)間。 
    unit - keepAliveTime參數(shù)的時(shí)間單位 
    workQueue - 用于在執(zhí)行任務(wù)之前使用的隊(duì)列。 這個(gè)隊(duì)列將僅保存execute方法提交的Runnable任務(wù)。 
    threadFactory - 執(zhí)行程序創(chuàng)建新線程時(shí)使用的工廠 
    handler - 執(zhí)行被阻止時(shí)使用的處理程序，因?yàn)檫_(dá)到線程限制和隊(duì)列容量 

我們這里的線程池類比一個(gè)公司,便于我們理解該類

• int maximumPoolSize,
  核心線程數(shù)(正式員工)
• maximumPoolSize
  池中允許的最大線程數(shù)(正式員工+臨時(shí)工)
• long keepAliveTime,
  多余的空閑線程的允許等待的最大時(shí)間(臨時(shí)工摸魚時(shí)間)
• TimeUnit unit,
  時(shí)間單位
  - BlockingQueue<Runnable> workQueue,
  任務(wù)隊(duì)列,該類中用一個(gè)submit方法,用于將任務(wù)注冊(cè)到線程池,加入到任務(wù)隊(duì)列中!
• ThreadFactory threadFactory,
  線程工廠,線程是如何創(chuàng)建的
• RejectedExecutionHandler handler
  拒絕策略
  但任務(wù)隊(duì)列滿了后怎么做
  1.阻塞等待,
  2.丟棄久任務(wù)
  3.忽略新任務(wù)
  …

可以看到java給我們提供的這個(gè)線程池類讓人頭大!
但是不必焦慮,我們只需要知道int maximumPoolSize,
核心線程數(shù)和 maximumPoolSize 池中允許的最大線程數(shù)即可!
面試問題
思考一個(gè)問題
我們有一個(gè)程序需要多線程并發(fā)處理一些任務(wù),使用線程池的話,需要設(shè)置多大的線程數(shù)?
這里的話,我們無法準(zhǔn)確的給出一個(gè)數(shù)值,我們要通過性能測(cè)試的方式找個(gè)一個(gè)平衡點(diǎn)!

例如我們寫一個(gè)服務(wù)器程序:服務(wù)器通過線程池多線程處理機(jī)用戶請(qǐng)求!如果要確定線程池的線程數(shù)的話,就需要通過對(duì)該服務(wù)器進(jìn)行性能分析,構(gòu)造很多很多請(qǐng)求模擬真實(shí)環(huán)境,根據(jù)這里不同的線程數(shù),來觀察處理任務(wù)的速度和當(dāng)個(gè)線程的cpu占用率!從而找到一個(gè)平衡點(diǎn)!
如果cpu暫用率過高,就無法應(yīng)對(duì)一些突發(fā)情況,服務(wù)器容易掛!

我們java根據(jù)上面的ThreadPoolExecutor類進(jìn)行封裝提供了一個(gè)簡(jiǎn)化版本的線程池!Executors供我們使用!
我們通過Executors的使用學(xué)習(xí),實(shí)現(xiàn)一個(gè)線程池!

Executors

java.util.concurrent.Executors

下面都是Executor類中創(chuàng)建線程池的一些靜態(tài)方法

創(chuàng)建可以擴(kuò)容的線程池

創(chuàng)建一個(gè)指定容量的線程池
創(chuàng)建單線程池
創(chuàng)建一個(gè)線程池含有任務(wù)隊(duì)列

我們重點(diǎn)學(xué)習(xí)創(chuàng)建指定大小得到線程池方法!

//Executors使用案例
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Demo3 {
    public static void main(String[] args) {
        //創(chuàng)建一個(gè)指定線程個(gè)數(shù)為10的線程池
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 100; i++) {
            int finalI = i;
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    System.out.println("hello executor!"+ finalI);
                }
            });
        }
    }
}

我們通過ExecutorService類中的submit可以將多個(gè)任務(wù)注冊(cè)到線程池中,然后線程池中的線程將任務(wù)并發(fā)執(zhí)行,大大提升了編程效率!可以看到,啪的一下,100個(gè)任務(wù)給10個(gè)線程一下就執(zhí)行結(jié)束了!

實(shí)現(xiàn)線程池

我們還是分析一下線程池用什么功能,里面都有些啥!

• 能夠描述任務(wù)(直接用runnable)
• 需要組織任務(wù)(使用BlockingQueue)
• 能夠描述工作線程
• 組織線程
• 需要實(shí)現(xiàn)往線程池里添加任務(wù)

//模擬實(shí)現(xiàn)線程池
class ThreadPool {
    //描述任務(wù) 直接使用Runnable
    //組織任務(wù)
    private BlockingQueue<Runnable> queue = new LinkedBlockingDeque<>();

    //描述工作線程
    static class Worker extends Thread {//繼承Thread類
        BlockingQueue<Runnable> queue = null;
        @Override
        public void run() {
            while (true){
                try {
                    //拿到任務(wù)
                    Runnable runnable = queue.take();
                    //執(zhí)行任務(wù)
                    runnable.run();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        //通過構(gòu)造方法拿到外面的任務(wù)隊(duì)列!
        public Worker(BlockingQueue<Runnable> queue) {
            this.queue = queue;
        }
    }

    //組織多個(gè)工作線程
    //將多個(gè)工作線程放入到workers中!
    public List<Thread>workers = new LinkedList<>();

    public ThreadPool(int n) {//指定放入線程數(shù)量
        for (int i = 0; i < n; i++) {//創(chuàng)建多個(gè)工作線程
            Worker worker = new Worker(queue);
            worker.start();//啟動(dòng)工作線程
            workers.add(worker);//放入線程池
        }
    }
    //創(chuàng)建一個(gè)方法供我們放入任務(wù)
    public void submit(Runnable runnable){
        try {
            queue.put(runnable);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

//測(cè)試代碼
public class demo5 {
    public static void main(String[] args) {
        //線程池線程數(shù)量10
        ThreadPool pool = new ThreadPool(10);
        for (int i = 0; i <100 ; i++) {//100個(gè)任務(wù)
            int finalI = i;
            pool.submit(new Runnable() {
                @Override
                public void run() {
                    System.out.println("hello ThreadPool "+ finalI);
                }
            });
        }
    }
}

運(yùn)行效果
文章來源地址http://www.zghlxwxcb.cn/news/detail-404832.html

案例總結(jié)

• 線程安全單例模式
• 阻塞隊(duì)列->生產(chǎn)著消費(fèi)者模型
• 定時(shí)器

1. MyTask類描述一個(gè)任務(wù) Runnable + time
2. 帶有優(yōu)先級(jí)的阻塞隊(duì)列
3. 掃描線程,不停從隊(duì)首取出元素,檢測(cè)時(shí)間是否到達(dá),并且執(zhí)行任務(wù),使用wait解決忙等位問題!
4. 實(shí)現(xiàn)schedule方法

• 線程池

1. 描述一個(gè)任務(wù)Runnable
2. 組織任務(wù),帶有優(yōu)先級(jí)的阻塞隊(duì)列
3. 創(chuàng)建一個(gè)工作線程work類,從任務(wù)隊(duì)列獲取任務(wù),執(zhí)行任務(wù)
4. 組織工作線程works數(shù)據(jù)結(jié)構(gòu)存放work
5. 實(shí)現(xiàn)一個(gè)submit方法將任務(wù)放入任務(wù)隊(duì)列中!

到了這里，關(guān)于多線程四大經(jīng)典案例及java多線程的實(shí)現(xiàn)的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！