快捷導航

Java中的阻塞隊列BlockingQueue使用詳解

更新時間：2023年10月24日 10:38:23 作者：皓亮君

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

1.BlockingQueue 簡介

BlockingQuene是一個阻塞隊列接口,當BlockingQueue操作無法立即響應時,有四種處理方式:

拋出異常;
返回特定的值,根據(jù)操作不同,可能是null或者false中的一個;
無限制的阻塞當前線程,直到操作可以成功為止;
根據(jù)阻塞超時設(shè)置來進行阻塞;
BlockingQueue的核心和未響應處理方式的對應形式如下:

方式	拋出異常	返回特定值	無限阻塞	超時
插入	add(e)	offer (e)	put(e)	offer(e,time,unit)
移除	remove()	poll()	take()	poll(time,unit)
查詢	element()	peek()

2.ArrayBlockingQueue(有界隊列)

ArrayBlockingQueue是基于數(shù)組實現(xiàn)的有界BlockingQueue,該隊列滿足先入先出(FIFO)的特性,當隊列滿時,存數(shù)據(jù)的操作會被阻塞;隊列空的時候,取數(shù)據(jù)的操作會被阻塞。

/**
 * @Author Dominick Li
 * @CreateTime 2022/3/6 20:03
 * @Description 消息生產(chǎn)者
 **/
public class Product implements Runnable {
    private BlockingQueue<String> bq;
    /**
     * 多少秒生產(chǎn)一條任務(wù)
     */
    private int period;
    private Random r = new Random();
    /**
     * 生產(chǎn)者名稱
     */
    private String name;
    Product(BlockingQueue<String> bq, int period, String name) {
        this.bq = bq;
        this.period=period;
        this.name=name;
    }
    @Override
    public void run() {
        try {
            while (true){
                Thread.sleep(period);
                String product=String.valueOf(r.nextInt(100));
                //如果隊列滿了則阻塞
                bq.put(product);
                System.out.println("生產(chǎn)者["+this.name+"]生產(chǎn)"+product+"，當前隊列中產(chǎn)品為:"+bq);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
/**
 * @Author Dominick Li
 * @CreateTime 2022/3/6 20:11
 * @Description 消費者
 **/
public class Cusumer implements Runnable {
    private BlockingQueue<String> bq;
    /**
     * 多少秒獲取一條任務(wù)
     */
    private int period;
    /**
     * 消費者名稱
     */
    private String name;
    Cusumer(BlockingQueue<String> bq, int period, String name) {
        this.bq = bq;
        this.period=period;
        this.name=name;
    }
    @Override
    public void run() {
        try {
            while (true){
                Thread.sleep(period);
                String value=bq.take();
                System.out.println("消費者["+this.name+"]消費"+value+"，當前隊列中產(chǎn)品為:"+bq);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}
public class Test {
    public static void main(String[] args) {
        BlockingQueue blockingQueue = new ArrayBlockingQueue(5);
        ExecutorService pool = Executors.newCachedThreadPool();
        pool.execute(new Product(blockingQueue, 1000, "生產(chǎn)者"));
        pool.execute(new Cusumer(blockingQueue, 5000, "消費者001"));
        pool.execute(new Cusumer(blockingQueue, 5000, "消費者002"));
        pool.shutdown();
    }
}

運行效果如下

在這里插入圖片描述

3.LinkedBlockingQueue(雙鎖線程安全隊列)

與ArrayBlockingQueue相比,LinkedBlockingQueue的重入鎖被分成了兩份,分別對應存值和取值,這種實現(xiàn)方法被稱為雙鎖隊列算法，這樣的好處是讀寫操作的lock操作由兩個鎖控制,因此可以同時進程讀操作和寫操作，這也是LinkedBlockingQueue吞吐量超出ArrayBlockingQueue的主要原因，但是使用兩個鎖比一個鎖復雜很多,需要考慮各種死鎖的狀態(tài)。使用方法和ArrayBlockingQueue一致

public class Test {
    public static void main(String[] args) {
        LinkedBlockingQueue linkedBlockingQueue = new LinkedBlockingQueue(5);
        ExecutorService pool = Executors.newCachedThreadPool();
        pool.execute(new Product(linkedBlockingQueue, 1000, "生產(chǎn)者"));
        pool.execute(new Cusumer(linkedBlockingQueue, 5000, "消費者1"));
        pool.shutdown();
    }
}

4.PriorityBlockingQueue(優(yōu)先級隊列)

優(yōu)先級阻塞隊列ProiorityBlockQueue不是FIFO(先入先出)隊列,它要求使用者提供一個Comparetor比較器,或者隊列內(nèi)部元素實現(xiàn)Comparable接口,隊頭元素會是整個隊列里的最小元素.

PriorityBlockQueue是用數(shù)組實現(xiàn)的最小堆結(jié)構(gòu),利用的原理是: 在數(shù)組實現(xiàn)的完全二叉樹中根節(jié)點的下標為子節(jié)點的下標除以2，長度是不定的,會隨著數(shù)據(jù)的增長而逐步擴容

public class PriorityProduct implements Comparable<PriorityProduct> {
    /**
     * 任務(wù)的優(yōu)先級
     */
    private int priority;
    private String productName;
    public PriorityProduct(int priority, String productName) {
        this.priority = priority;
        this.productName = productName;
    }
    @Override
    public int compareTo(PriorityProduct o) {
        if (o == null) return -1;
        if (o == this) return 0;
        return o.priority - this.priority;
    }
    @Override
    public String toString(){
        return "{priority="+priority+",name="+this.productName;
    }
}
public class PriorityBlockQueueProduct implements Runnable {
    private PriorityBlockingQueue<PriorityProduct> bq;
    /**
     * 多少秒生產(chǎn)一條任務(wù)
     */
    private int period;
    private Random r = new Random();
    public PriorityBlockQueueProduct(PriorityBlockingQueue<PriorityProduct> bq, int period) {
        this.bq = bq;
        this.period = period;
    }
    @Override
    public void run() {
        try {
            while (true) {
                Thread.sleep(period);
                if(bq.size()>10){
                    //限制大小
                    continue;
                }
                  //隨機生成優(yōu)先級5以內(nèi)的
                PriorityProduct priorityProduct = new PriorityProduct(r.nextInt(5), "test");
                //如果隊列滿了則阻塞
                bq.put(priorityProduct);
                //System.out.println("生產(chǎn)者商品[" +priorityProduct + "]，當前隊列中產(chǎn)品為:" + bq);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
public class PriorityBlockQueueCusumer implements Runnable {
    private PriorityBlockingQueue<PriorityProduct> bq;
    /**
     * 多少秒消費一條任務(wù)
     */
    private int period;
    public PriorityBlockQueueCusumer(PriorityBlockingQueue<PriorityProduct> bq, int period) {
        this.bq = bq;
        this.period = period;
    }
    @Override
    public void run() {
        try {
            while (true) {
                Thread.sleep(period);
                //如果隊列滿了則阻塞
                PriorityProduct priorityProduct=bq.take();
                System.out.println("消費產(chǎn)品[" +priorityProduct + "]，當前隊列中產(chǎn)品為:" + bq);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
public class PriorityTest {
    public static void main(String[] args) {
        PriorityBlockingQueue<PriorityProduct> priorityProducts=new PriorityBlockingQueue<>();
        ExecutorService executorService= Executors.newFixedThreadPool(2);
        executorService.execute(new PriorityBlockQueueProduct(priorityProducts,100));
        executorService.execute(new PriorityBlockQueueCusumer(priorityProducts,1000));
    }
}

運行結(jié)果如下,可以查看消費者在消費的時候只會消費任務(wù)隊列中優(yōu)先級最高的任務(wù)

在這里插入圖片描述