快捷導(dǎo)航

java之阻塞隊(duì)列BlockingQueue解析

更新時(shí)間：2023年10月24日 10:18:11 作者：強(qiáng)強(qiáng)愛(ài)java

這篇文章主要介紹了java之阻塞隊(duì)列BlockingQueue解析,阻塞隊(duì)列是一種線(xiàn)程安全的數(shù)據(jù)結(jié)構(gòu),用于在多線(xiàn)程環(huán)境下進(jìn)行數(shù)據(jù)交換,它提供了一種阻塞的機(jī)制,當(dāng)隊(duì)列為空時(shí),消費(fèi)者線(xiàn)程將被阻塞,直到隊(duì)列中有數(shù)據(jù)可供消費(fèi),需要的朋友可以參考下

一、阻塞隊(duì)列概念

阻塞隊(duì)列，顧名思義，首先它是一個(gè)隊(duì)列（先進(jìn)先出），而一個(gè)阻塞隊(duì)列在數(shù)據(jù)結(jié)構(gòu)所起到的作用大致如下圖：

在這里插入圖片描述

線(xiàn)程1往阻塞隊(duì)列中添加元素，而線(xiàn)程2從阻塞隊(duì)列中移除元素
當(dāng)阻塞隊(duì)列是空是，從隊(duì)列中獲取元素的操作會(huì)被阻塞
當(dāng)阻塞隊(duì)列是滿(mǎn)時(shí)，從隊(duì)列中添加元素的操作會(huì)被阻塞
試圖從空的阻塞隊(duì)列中獲取元素的線(xiàn)程將會(huì)被阻塞，直到其他的線(xiàn)程往空的隊(duì)列插入新的元素。
試圖網(wǎng)已滿(mǎn)的阻塞隊(duì)列中添加新元素的線(xiàn)程同樣會(huì)被阻塞，直到其他的線(xiàn)程從列中移除一個(gè)或者多個(gè)元素或者完全清空隊(duì)列后使隊(duì)列重新變得空閑起來(lái)并后續(xù)新增

二、阻塞隊(duì)列的好處

1.在多線(xiàn)程領(lǐng)域：所謂阻塞，在某些情況下會(huì)掛起線(xiàn)程，一旦滿(mǎn)足條件，被掛起的線(xiàn)程又會(huì)自動(dòng)被喚醒。

2.我們不需要關(guān)心什么時(shí)候需要阻塞線(xiàn)程，什么時(shí)候需要喚醒線(xiàn)程，因?yàn)檫@一切BlockingQueue（阻塞隊(duì)列）都給你一手包辦了

PS：在concurrent包發(fā)布以前，在多線(xiàn)程環(huán)境下，我們每個(gè)程序員都必須自己控制這些細(xì)節(jié)，尤其還要兼顧效率和線(xiàn)程安全，而這回給我們程序帶來(lái)不小的復(fù)雜度

三、阻塞隊(duì)列種類(lèi)

ArrayBlockingQueue：由數(shù)據(jù)結(jié)構(gòu)組成的有界阻塞隊(duì)列
LinkedBlockingQueue：由鏈表結(jié)構(gòu)組成的有界（但大小默認(rèn)值為 Integer.MAX_VALUE )阻塞隊(duì)列
PriorityBlockingQueue：支持優(yōu)先級(jí)排序的無(wú)界阻塞隊(duì)列
DelayQueue：使用優(yōu)先級(jí)隊(duì)列實(shí)現(xiàn)的延遲無(wú)界阻塞隊(duì)列
SynchronousQueue：不存儲(chǔ)元素的阻塞隊(duì)列，也即單個(gè)元素的隊(duì)列
LinkedTransferQueue：由鏈表結(jié)構(gòu)組成的無(wú)界阻塞隊(duì)列
LinkedBlockingDeque：由歷覽表結(jié)構(gòu)組成的雙向阻塞隊(duì)列

PS：重點(diǎn)掌握ArrayBlockingQueue、LinkedBlockingQueue、SychronousQueue三種

四、BlockingQueue的核心方法

方法類(lèi)型	拋出異常	特殊值	一直阻塞	超時(shí)退出
插入	add(e)	offer(e)	put(e)	offer(e,time,unit)
移除	remove()	poll()	take	poll(time,unit)
檢查	element()	peek()	不可用	不可用

說(shuō)明：

方法類(lèi)型	情況
拋出異常	當(dāng)阻塞隊(duì)列滿(mǎn)時(shí)，再往隊(duì)列中add會(huì)拋 IllegalStateException: Queue full；當(dāng)阻塞隊(duì)列空時(shí)，再?gòu)年?duì)列里remove會(huì)拋 NoSuchElementException
特殊值	offer(e)插入方法，成功true失敗false poll() 移除方法，成功返回出隊(duì)列的元素，隊(duì)列里沒(méi)有就返回null
一直阻塞	當(dāng)阻塞隊(duì)列滿(mǎn)時(shí)，生產(chǎn)者線(xiàn)程繼續(xù)往隊(duì)列里put元素，隊(duì)列會(huì)一直阻塞線(xiàn)程直到take數(shù)據(jù)或響應(yīng)中斷退出當(dāng)阻塞隊(duì)列空時(shí)，消費(fèi)者線(xiàn)程試圖從隊(duì)列take元素，隊(duì)列會(huì)一直阻塞消費(fèi)者線(xiàn)程直到隊(duì)列可用
超時(shí)退出	當(dāng)阻塞隊(duì)列滿(mǎn)時(shí)，隊(duì)列會(huì)阻塞生產(chǎn)者線(xiàn)程一定時(shí)間，超過(guò)限時(shí)后生產(chǎn)者線(xiàn)程會(huì)退出

五、示例代碼

1.使用SychronousQueue隊(duì)列

package com.jian8.juc.queue;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;

/*
 * 阻塞隊(duì)列SynchronousQueue演示
 * */
public class SynchronousQueueDemo {
    public static void main(String[] args) {
        BlockingQueue<String> blockingQueue = new SynchronousQueue<>();

        new Thread(() -> {
            try {
                System.out.println(Thread.currentThread().getName() + "\t put 1");
                blockingQueue.put("1");
                System.out.println(Thread.currentThread().getName() + "\t put 2");
                blockingQueue.put("2");
                System.out.println(Thread.currentThread().getName() + "\t put 3");
                blockingQueue.put("3");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "AAA").start();

        new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName() + "\t take" + blockingQueue.take());

                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName() + "\t take" + blockingQueue.take());

                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName() + "\t take" + blockingQueue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "BBB").start();
    }
}

運(yùn)行結(jié)果：

在這里插入圖片描述

2.傳統(tǒng)版生產(chǎn)者消費(fèi)者模式

package com.jian8.juc.queue;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 一個(gè)初始值為零的變量，兩個(gè)線(xiàn)程對(duì)其交替操作，一個(gè)加1一個(gè)減1，來(lái)5輪
 * 1. 線(xiàn)程  操作  資源類(lèi)
 * 2. 判斷  干活  通知
 * 3. 防止虛假喚起機(jī)制
 */
public class ProdConsumer_TraditionDemo {
    public static void main(String[] args) {
        ShareData shareData = new ShareData();
        for (int i = 1; i <= 5; i++) {
            new Thread(() -> {
                try {
                    shareData.increment();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }, "ProductorA " + i).start();
        }

        for (int i = 1; i <= 5; i++) {
            new Thread(() -> {
                try {
                    shareData.decrement();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }, "ConsumerA  " + i).start();
        }

        for (int i = 1; i <= 5; i++) {
            new Thread(() -> {
                try {
                    shareData.increment();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }, "ProductorB " + i).start();
        }

        for (int i = 1; i <= 5; i++) {
            new Thread(() -> {
                try {
                    shareData.decrement();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }, "ConsumerB  " + i).start();
        }

    }
}

//資源類(lèi)
class ShareData {
    private int number = 0;
    private Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();

    //生產(chǎn)
    public void increment() throws Exception {
        lock.lock();
        try {
            //1.判斷
            while (number != 0) {
                //等待中，不能生產(chǎn)
                condition.await();
            }
            //2.干活
            number++;
            System.out.println(Thread.currentThread().getName() + "\t" + number);
            //3.通知
            condition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    //消費(fèi)
    public void decrement() throws Exception {
        lock.lock();
        try {
            //1.判斷
            while (number == 0) {
                //等待中，不能消費(fèi)
                condition.await();
            }
            //2.消費(fèi)
            number--;
            System.out.println(Thread.currentThread().getName() + "\t" + number);
            //3.通知
            condition.signalAll();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

}

運(yùn)行結(jié)果（部分）：

在這里插入圖片描述

3.阻塞隊(duì)列版（以ArrayBlockingQueue為例）生產(chǎn)者消費(fèi)者模式

package com.jian8.juc.queue;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class ProdConsumer_BlockQueueDemo {
    public static void main(String[] args) {
        MyResource myResource = new MyResource(new ArrayBlockingQueue<>(10));

        new Thread(() -> {
            System.out.println(Thread.currentThread().getName() + "\t生產(chǎn)線(xiàn)程啟動(dòng)");
            try {
                myResource.myProd();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "Prod").start();

        new Thread(() -> {
            System.out.println(Thread.currentThread().getName() + "\t消費(fèi)線(xiàn)程啟動(dòng)");
            try {
                myResource.myConsumer();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "Consumer").start();

        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("5s后main叫停，線(xiàn)程結(jié)束");

        try {
            myResource.stop();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

//資源類(lèi)
class MyResource {
    private volatile boolean flag = true;//默認(rèn)開(kāi)啟，進(jìn)行“生產(chǎn)+消費(fèi)”活動(dòng)
    private AtomicInteger atomicInteger = new AtomicInteger();
    BlockingQueue<String> blockingQueue = null;//消息隊(duì)列

    public MyResource(BlockingQueue<String> blockingQueue) {
        this.blockingQueue = blockingQueue;
        System.out.println(blockingQueue.getClass().getName());
    }

    //生產(chǎn)
    public void myProd() throws Exception {
        String data = null;
        boolean retValue;
        while (flag) {
            data = atomicInteger.incrementAndGet() + "";
            //超過(guò)2s沒(méi)生產(chǎn)成功，退出生產(chǎn)
            retValue = blockingQueue.offer(data, 2, TimeUnit.SECONDS);
            if (retValue) {
                System.out.println(Thread.currentThread().getName() + "\t插入隊(duì)列" + data + "成功");
            } else {
                System.out.println(Thread.currentThread().getName() + "\t插入隊(duì)列" + data + "失敗");
            }
            TimeUnit.SECONDS.sleep(1);
        }
        System.out.println(Thread.currentThread().getName() + "\t大老板叫停了，flag=false，生產(chǎn)結(jié)束");
    }

    //消費(fèi)
    public void myConsumer() throws Exception {
        String result = null;
        while (flag) {
            //超過(guò)2s沒(méi)從隊(duì)列獲取數(shù)據(jù)，消費(fèi)退出
            result = blockingQueue.poll(2, TimeUnit.SECONDS);
            if (null == result || result.equalsIgnoreCase("")) {
                flag = false;
                System.out.println(Thread.currentThread().getName() + "\t超過(guò)2s沒(méi)有取到蛋糕，消費(fèi)退出");
                System.out.println();
                return;
            }
            System.out.println(Thread.currentThread().getName() + "\t消費(fèi)隊(duì)列" + result + "成功");
        }
    }

    public void stop() throws Exception {
        flag = false;
    }
}

運(yùn)行結(jié)果：

在這里插入圖片描述