阻塞队列_线程安全版本_生产消费者模型

用户10788736

发布于 2023-10-16 08:40:20

1350

发布于 2023-10-16 08:40:20

前言

在前面PriorityQueue优先级队列_Y君的进化史的博客-CSDN博客，我们学习了优先级队列，但是发现，当一个线程将优先级队列使用完之后，会自动退出程序，如果此时我们想使其一直等待到下一个任务的录入，就需要通过阻塞的功能实现，于是就有了阻塞队列的诞生。

一、阻塞队列原理

当队列满时，继续入队列就会产生阻塞，直到其他线程从队列中取出元素为止；
当队列空时，继续出队列就会产生阻塞，直到其他线程往队列中添加元素为止。

Java标准库对于阻塞队列的实现：

BlockingQueue<Integer> queue1 = new LinkedBlockingQueue<>();
BlockingQueue<Integer> queue2 = new PriorityBlockingQueue<>();
BlockingQueue<Integer> queue3 = new ArrayBlockingQueue<>(12);//12 是指任务的最多数目

对于BlockingQueue,offer 和 poll 是不带阻塞功能的（因为BlockingQueue 继承 Queue），put 和 take 带阻塞功能的，这里的阻塞功能是使用wait-notify实现的。

二、生产消费者模型

（一）解耦合

通过引入阻塞队列，介绍了各处理器之间的联系，哪怕哪个机子先报废了，也不必担心其他机子受到太多的影响，起到了解耦合的作用。

（二）削峰填谷

原来的处理机制是，A传送一个信号，B就接收一个信号。若A服务器配置更好，A能承受的信号多，B能承受的信号少，该机制就会挂机。

但是通过阻塞队列的引入，不管A传送的信号是多还是少，B依旧能以自身的节奏去处理信号，起到了削峰填谷的作用。

三、代码实现阻塞队列

//阻塞队列
class MyBlockingQueue{
    private String[] elem = new String[10];
    //头
    volatile private int head;
    //尾
    volatile private int end;
 
    //锁
    Object lock = new Object();
    //插入
    public void put(String tmp) throws InterruptedException {
        synchronized (lock){
            while((end+1)%elem.length == head){
                //队列满了，阻塞
                lock.wait();
            }
            elem[end] = tmp;
            end++;
            if(end >= elem.length){
                end = 0;
            }
            lock.notify();
        }
    }
 
    //出队列
    public String take() throws InterruptedException {
        synchronized (lock){
            while(end == head){
                //队列空的，出队列阻塞
                lock.wait();
            }
            String tmp = elem[head];
            head++;
            if(head >= elem.length){
                head = 0;
            }
            lock.notify();
            return tmp;
        }
    }
 
}

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2023-08-11，如有侵权请联系 cloudcommunity@tencent.com 删除

博客