Java多線程(單例模式,堵塞隊列,定時器)詳解

更新時間：2021年08月16日 09:18:27 作者：caiyec

這篇文章主要介紹了java多線程的(單例模式,堵塞隊列,定時器)，具有一定參考價值，加深多線程編程的理解還是很有幫助的，需要的朋友可以參考下

一、單例模式

單例模式是一種設(shè)計模式，針對一些特定的場景，研究出對應(yīng)的解決方案，。有些對象在代碼中只應(yīng)該有一個實例，單例模式就是強(qiáng)制某個類只能有一個實例。

單例模式的實現(xiàn)，主要依托于static關(guān)鍵字（被static 修飾的成員，靜態(tài)成員，把當(dāng)前的成員變成類屬性而不是實例屬性~）每個類對象只有一份

單例模式實現(xiàn)有兩種，餓漢模式和懶漢模式

餓漢模式

餓漢模式實現(xiàn)：實例創(chuàng)建出現(xiàn)在“類加載”階段（第一次使用到這個類的時候，就會把這個類.class加載到內(nèi)存里），線程安全

public class TestSinger {
    //實現(xiàn)單例模式
    static class Singleton{
        //創(chuàng)建一個成員，保存唯一的一個Singleton實例
        private static Singleton instance=new Singleton();
        //提供方法獲取實例
        public static Singleton getInstance(){
            return instance;
        }
        private Singleton(){
        }
    }
    public static void main(String[] args) {
        //獲取到一個實例 ，只能通過 getInstance 無法通過new 的方式來創(chuàng)建新的Singleton
        Singleton s=Singleton.getInstance();
    }
}

懶漢模式

第一次調(diào)用getInstance 方法創(chuàng)建實例（線程不安全）

public class TestSingleton {
    //懶漢模式
    //創(chuàng)建實例的時機(jī)是第一次調(diào)用時創(chuàng)建，比餓漢模式更遲
    static class Singleton{
        private static Singleton instance=null;
        public static Singleton getInstance(){
            if(instance==null){
                instance=new Singleton();
            }
            return instance;
        }
        private Singleton(){
        }
    }
    public static void main(String[] args) {
        Singleton s=new Singleton();
    }
}

一般來說懶漢模式更好（但不絕對），懶漢模式更高效，但是餓漢模式是線程安全的，懶漢模式是存在線程不安全的狀況，因為懶漢模式有創(chuàng)建線程實例操作，此操作不是原子性，

  public static Singleton getInstance(){
            if(instance==null){
                instance=new Singleton();
            }
            return instance;
        }

懶漢模式這里操作先進(jìn)行讀操作（LOAD）,之后進(jìn)行比較CMP 之后NEW SAVE（寫入內(nèi)存），如果這里有兩個線程執(zhí)行，會發(fā)生搶占式,因為這里操作不是原子性的，所有會發(fā)生創(chuàng)建多個實例的情況，出現(xiàn)了BUG，

在這里插入圖片描述

這里我們通過加鎖操作來使得操作變?yōu)樵有?，使得懶漢模式變?yōu)榫€程安全的，可以把鎖加到方法上，這時候是針對CMP,NEW 和 SAVE 操作都進(jìn)行了加鎖，三個操作都是串行的，但是這種效率太低了，我們應(yīng)該把鎖作用范圍更小一點(diǎn)，針對CMP（判斷）和NEW 操作進(jìn)行加鎖，SAVE 只是讀操作，并沒有修改，不需要加鎖，提高效率。

public static Singleton getInstance(){
       synchronized (Singleton.class){
           if(instance==null){
                instance=new Singleton();
             }
        }
     return instance;
	}

但是這樣的代碼，符出的代價太大了，因為每次調(diào)用都會進(jìn)行加鎖，我們只是需要instance未初始化之前，才涉及到線程安全問題，后續(xù)已經(jīng)初始化了，就每次要每次都執(zhí)行加鎖，而是只是進(jìn)行判斷就好了，所以又修改了代碼，改為雙if判斷

public static Singleton getInstance(){
           if(instance==null){
               synchronized (Singleton.class){
                   if(instance==null){
                       instance=new Singleton();
                   }
               }
           }
            return instance;
        }

但是這樣寫還是會有瑕疵，因為在多線程的情況下，可能多個線程進(jìn)行讀操作，由于編譯器優(yōu)化，可能在寄存器讀取，而這時候執(zhí)行操作還沒有執(zhí)行完，還是null的狀態(tài)，所以我們也要在獲取實例時候加上鎖

懶漢模式

保證線程安全：

1.加鎖，把if判斷和new操作加鎖

2.雙重if循環(huán)

3.volatile 關(guān)鍵字

    //懶漢模式
    static class Singleton{
        volatile  private static Singleton instance=null;
        public static Singleton getInstance(){
           if(instance==null){
               synchronized (Singleton.class){
                   if(instance==null){
                       instance=new Singleton();
                   }
               }
           }
            return instance;
        }
        private Singleton(){
        }
    }
    public static void main(String[] args) {
        Singleton s=new Singleton();
    }

針對單例模式的線程安全要點(diǎn)：

1）加鎖（在合適的位置加鎖，CMP（判斷）和NEW（創(chuàng)建）時加鎖，同時加鎖的范圍也不能太大，避免降低效率）

2）雙重判斷（保證需要加鎖時候才加鎖，一旦初始化完畢了，就不用創(chuàng)建實例，都為讀操作，就沒必要加鎖了）

3）volatile 保證外層 if 讀操作，讀到的數(shù)值都是最新的，不會出現(xiàn)一個正在創(chuàng)建實例，而讀取時是NULL 進(jìn)入IF判斷的情況

二、堵塞隊列

堵塞隊列是什么？一種線程安全的隊列，

1.首先堵塞隊列是線程安全的（內(nèi)部實現(xiàn)了加鎖控制），
2.當(dāng)隊列滿的時候，此時就會堵塞，一直到堵塞隊列不滿的情況下才會完成插入，當(dāng)隊列為空時，從隊列中取元素時，也會發(fā)生堵塞。

堵塞隊列的作用：

幫助我們完成“生產(chǎn)者消費(fèi)者模型”，作用于服務(wù)器開發(fā)

生產(chǎn)者和消費(fèi)者模型通過某種交易場所（某數(shù)據(jù)結(jié)構(gòu)）來進(jìn)行交互，堵塞隊列就是其中的一種數(shù)據(jù)結(jié)構(gòu)，能夠很好的協(xié)調(diào)生產(chǎn)者和消費(fèi)者之間的關(guān)系，

實際案例（服務(wù)器請求）：

一個服務(wù)器，同一時刻可能收到很多請求，但是服務(wù)器處理能力是有限的，如果同一時間服務(wù)器收到的請求太多了，服務(wù)器可能就掛了…，針對這樣的場景，使用生產(chǎn)者和消費(fèi)者模式來進(jìn)行“削峰”，削弱請求峰值對服務(wù)器的沖擊力，如果服務(wù)器面對請求太多了，實際上先把請求放入堵塞隊列中，應(yīng)用程序按照固定的結(jié)構(gòu)從堵塞隊列中取出，這些請求沖擊的是堵塞隊列本身，請求在這里耗著，不會消耗太多的CPU資源，緩解服務(wù)器壓力

消息隊列，是堵塞隊列的上級

1.消息隊列中數(shù)據(jù)是有類型的（topic），按照topic進(jìn)行分類，把相同topic的數(shù)據(jù)放到不同的隊伍中，分別進(jìn)行排隊，一個消息隊列，可以支撐多個業(yè)務(wù)的多組數(shù)據(jù)~~

2.消息隊列往往是單獨(dú)的服務(wù)器/服務(wù)器集群，通過網(wǎng)絡(luò)通信的方式，進(jìn)行生產(chǎn)者和消費(fèi)者模型

3.還支持持久化存儲（數(shù)據(jù)存儲在磁盤上）

4.消費(fèi)的時候支持多種消費(fèi)模式

a)指定位置消費(fèi)（不一定只是取出隊首元素）

b)鏡像模式消費(fèi)（一個數(shù)據(jù)可以被取多次，不是取一次直接刪除）

實現(xiàn)堵塞隊列：

public static void main(String[] args) {
        //BlockingDeque 本身是一個interface 不能去new
        BlockingDeque<String> blockingDeque=new LinkedBlockingDeque<>();
        try {
            //put 和 take 都有堵塞功能
            //堵塞隊列也有普通方法但是沒有堵塞功能。
            blockingDeque.put("hello");
            String elem=blockingDeque.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

實現(xiàn)一個生產(chǎn)者和消費(fèi)者模型

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
public class Demo2 {
    //實現(xiàn)生產(chǎn)者和消費(fèi)者模型
    public static void main(String[] args) {
        BlockingDeque<String> queue=new LinkedBlockingDeque();
        //創(chuàng)建生產(chǎn)者線程
        Thread producer=new Thread(){
            @Override
            public void run() {
                for(int i=0;i<10000;i++){
                    try {
                        System.out.println("producer 生成 str"+i);
                        queue.put("str "+i);
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        producer.start();
        //消費(fèi)者線程
        Thread customer=new Thread(){
            @Override
            public void run() {
                while(true){
                    try {
                        String elem=queue.take();
                        System.out.println("customer 獲取到" + elem);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        customer.start();
        try {
            producer.join();
            customer.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

這里實現(xiàn)的是生產(chǎn)者每一秒生成一個，生產(chǎn)者比消費(fèi)者慢

在這里插入圖片描述

可以借助堵塞隊列的最大長度來設(shè)置一個生產(chǎn)者比消費(fèi)者快的情況，將最大長度設(shè)為10，使用sleep 一秒消費(fèi)一個，但是一直在生產(chǎn)，這樣就是生產(chǎn)者大于消費(fèi)者，主要使用put（）和take（）方法來操作堵塞隊列

實現(xiàn)BlockingQueue

1）首先要實現(xiàn)一個隊列，可以用鏈表或者數(shù)組實現(xiàn)隊列，這里使用數(shù)組實現(xiàn)一個隊列（環(huán)形隊列），定義兩個變量head,tail來標(biāo)記數(shù)組頭部和尾部，插入元素時，插在tail位置，tail++,出隊列時取出head位置元素，head++，定義一個變量來標(biāo)記長度，如果長度等于數(shù)組長度，則要回到數(shù)組的頭部，來實現(xiàn)環(huán)形數(shù)組

public class ThreadDemo1 {
    //自己實現(xiàn)堵塞隊列，先通過數(shù)組實現(xiàn)普通隊列
    static class BlockingQueue{
        private int[] array=new int[1000];
        private int head=0;//記錄頭部
        private int tail=0;//記錄尾部
        private int size=0;
        //實現(xiàn)入隊列
        public void put(int value){
            if(size==array.length){
                System.out.println("隊列滿了，不能插入");
                return ;
            }
            array[tail]=value;
            tail++;
            //解決環(huán)形數(shù)組
            if(tail>=array.length){
                tail=0;
            }
            size++;
        }
        //實現(xiàn)出隊列
        public Integer take(){
            if(size==0){
                return null;
            }
            int ret=array[head];
            head++;
            if(head>=array.length){
                head=0;
            }
            size--;
            return ret;
        }
    }
}

2.為了保證線程安全給隊列進(jìn)行加鎖操作，并且實現(xiàn)堵塞隊列

注意實現(xiàn)堵塞隊列，此時隊列是滿的，多個線程實現(xiàn)都是要等待，當(dāng)一個線程取走一個元素，就會通知其他線程隊列不滿，多個線程就要競爭鎖，所以獲取到鎖操作后，還是要判斷隊列是否滿，可能這個線程沒有競爭到鎖，所以要用while()來進(jìn)行等待

static class BlockingQueue{
        private int[] array=new int[1000];
        private int head=0;//記錄頭部
        private int tail=0;//記錄尾部
        //記錄隊列中元素長度
        private int size=0;
        //引入一個鎖對象
        private Object locker=new Object();
        //實現(xiàn)入隊列
        public void put(int value) throws InterruptedException {
            synchronized (locker){
                while(size==array.length){
                    locker.wait();
                }
                array[tail]=value;
                tail++;
                //解決環(huán)形數(shù)組
                if(tail>=array.length){
                    tail=0;
                }
                size++;
                locker.notifyAll();
            }
        }
        //實現(xiàn)出隊列
        public Integer take() throws InterruptedException {
            int ret=0;
            synchronized (locker){
                while (size==0){
                    locker.wait();
                }
                ret=array[head];
                head++;
                if(head>=array.length){
                    head=0;
                }
                size--;
                locker.notifyAll();//喚醒操作，提醒等待元素，隊列有位置了
            }
            return ret;
        }
    }

創(chuàng)建一個生產(chǎn)者消費(fèi)者模型來檢驗自己實現(xiàn)的堵塞隊列是否成功

public static void main(String[] args) throws InterruptedException {
        BlockingQueue queue=new BlockingQueue();
        Thread producer=new Thread(){
            @Override
            public void run() {
                for(int i=0;i<10000;i++){
                    try {
                        System.out.println("生產(chǎn)了元素："+ i);
                        queue.put(i);
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        };
        producer.start();
        Thread customer=new Thread(){
            @Override
            public void run() {
                try {
                    while(true){
                        int ret=queue.take();
                        System.out.println("消費(fèi)了元素 "+ ret);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        customer.start();
        producer.join();
        customer.join();
    }

在這里插入圖片描述

實現(xiàn)了一個簡單的堵塞隊列

三、定時器

定時器就是鬧鐘，給定時器設(shè)定一個任務(wù)，約定某個任務(wù)XXX時間后執(zhí)行

目的：讓某個任務(wù)在某個時間點(diǎn)執(zhí)行，不是立刻執(zhí)行

使用Timer 提供的核心接口 schedule 指定一個任務(wù)交給定時器，再一定的時間之后執(zhí)行這個任務(wù)

實現(xiàn)定時器
1）Timer 類中要包含一個Task類，每個Task類就表示一個具體的任務(wù)，Task里面包含一個時間戳（啥時候執(zhí)行這個任務(wù)），還包含了一個Runnable 實例（用來表示具體任務(wù)是啥）
2）Timer里面通過一個帶優(yōu)先級的堵塞隊列，來組織若干個task，根據(jù)時間先后來排優(yōu)先級，快帶時間的任務(wù)優(yōu)先級更高
3）Timer 中還需要一個專門的線程，讓這個線程不停掃描隊首元素，看看隊首元素是不是可以執(zhí)行了，如果可以執(zhí)行了，就執(zhí)行這個任務(wù)，如果不能執(zhí)行，就繼續(xù)在隊列中等待。

實現(xiàn)定時器:

import java.util.concurrent.PriorityBlockingQueue;
public class ThreadDemo2 {
    //實現(xiàn)一個簡單的定時器  task要放到一個優(yōu)先隊列中，但是優(yōu)先隊列中需要進(jìn)行比較排序
    static class Task implements Comparable<Task>{
        //啥時候去執(zhí)行
        private long time;
        //執(zhí)行什么
        private Runnable command;
        //一般去設(shè)定定時器的時候，傳入的時間，一般都是時間間隔
        public Task(Runnable command,long time){
            this.command=command;
            //記錄絕對時間
            this.time=System.currentTimeMillis()+time;
        }
        public void run(){
            command.run();
        }
        @Override
        public int compareTo(Task o) {
        //時間較小的排在前面
            return (int)(this.time-o.time);
        }
    }
    static class Timer{
        //創(chuàng)建一個帶優(yōu)先級的堵塞隊列
       private PriorityBlockingQueue<Task> queue=new PriorityBlockingQueue<>();
       //使用這個對象來實現(xiàn)線程之間的協(xié)調(diào)任務(wù)
        private Object mailBox=new Object();
        //schedule 方法的功能就是把一個Task 放到Timer中
        public void schedule(Runnable command,long after){
            Task task=new Task(command,after);
            queue.put(task);
            //當(dāng)worker 線程中包含wait 機(jī)制的時候，在安排任務(wù)的時候就需要顯式的喚醒一下了
            synchronized (mailBox){
                mailBox.notify();
            }
        }
        public Timer(){
            //創(chuàng)建一個線程，讓這個線程去掃描隊列的隊首元素
            Thread worker=new Thread(){
                @Override
                public void run() {
                    while (true){
                        //取出隊首元素，判定一下這個元素能不能執(zhí)行
                        try {
                            Task task=queue.take();
                            long currentTime=System.currentTimeMillis();
                            if(currentTime>=task.time){
                                //時間到了執(zhí)行任務(wù)
                                task.run();
                            }else{
                                //時間沒到，繼續(xù)等待
                                queue.put(task);
                                synchronized (mailBox){
                                    mailBox.wait(task.time-currentTime);
                                }
                            }
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            };
            worker.start();
        }
    }
}