• 如果您觉得本站非常有看点,那么赶紧使用Ctrl+D 收藏吧

并发编程核心——九、阻塞队列

互联网 diligentman 6天前 6次浏览

使用LOCK.notify,LOCK.wait模拟一个阻塞队列.

阻塞队列也是一个队列,拥有队列某个时刻拥有元素数量,队列最大容纳元素数量,队列最小数量,队列这些属性。
阻塞的含义是如果队列中元素个数与队列最大容纳元素数量相同,那么再放元素,放元素的线程会进入wait队列的阻塞状态,等待其他线程唤醒后才能放元素;如果队列中的元素个数与最小容纳元素数量相同,那么取数据的线程会进入wait队列的阻塞状态,等待其他线程唤醒后才能取数据。
增加属性:锁
操作:构造方法,私有属性的公有获取方法,获取队首元素的方法(加锁),获取队尾元素数据(加锁)

代码1:自己封装的阻塞队列

package com.bjsxt.chapter09;

import java.util.LinkedList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 模拟阻塞队列
 * @author wmr
 * @date 2021/2/20
 */
public class MyQueue {
    // 队列
    private LinkedList<Object> list = new LinkedList<>();
    // 元素个数
    private AtomicLong count = new AtomicLong(0);
    // 最大数量
    private final long maxSize;
    // 最小数量
    private final long minSize = 0;
    // 锁
    private final Object LOCK = new Object();
    // 构造方法
    public MyQueue(long maxSize) {
        this.maxSize = maxSize;
    }
    // 获取元素个数(封装变量的存取操作到方法中,这样便于维护变量的存取增加的操作)
    public long getCount(){
        return this.count.get();
    }
    // 获取队列最多能放元素数量
    public long getMaxSize(){
        return this.maxSize;
    }
    // 获取队列最少能放元素数量
    public long getMinSize(){
        return this.minSize;
    }
    // 入队
    public void put(Object o){
        synchronized (LOCK){
            // 队列满,入队线程进入wait队列的阻塞状态
            if(this.count.get() == this.maxSize){
                try{
                    LOCK.wait();
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
            // 否则,入队,数量加一,唤醒其他线程执行出队操作
            this.list.add(o);
            this.count.incrementAndGet();
            LOCK.notify();
            try{
                TimeUnit.SECONDS.sleep(2);
            }catch (InterruptedException e){
                e.printStackTrace();
            }

        }
    }
    // 出队
    public Object take(){
        Object ret = null;
        synchronized (LOCK){
            // 队列空,出队线程进入wait队列的阻塞状态
            if(this.count.get() == this.minSize){
                try{
                    LOCK.wait();
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
            // 否则,返回第一个元素,删除第一个元素,数量减一,唤醒其他线程进行入队操作
            ret = this.list.getFirst();
            this.list.removeFirst();
            this.count.decrementAndGet();
            LOCK.notify();
            try{
                TimeUnit.SECONDS.sleep(2);
            }catch (InterruptedException e){
                e.printStackTrace();
            }
        }
        return ret;
    }
}

代码2:程序员业务操作的类,测试阻塞队列的使用

package com.bjsxt.chapter09;


import java.util.concurrent.TimeUnit;

/**
 * @author wmr
 * @date 2021/2/20
 */
public class Demo {
    public static void main(String[] args) throws InterruptedException {
        System.out.println("-----------main------------");
        // 模拟程序员在外面业务代码中使用阻塞队列,调用构造器创建一个阻塞队列
        final MyQueue myQueue = new MyQueue(5);
        // 向阻塞队列中依次放进"a","b","c","d","e"五个元素
        myQueue.put("a");
        System.out.println("元素" + "a" + "入队");
        myQueue.put("b");
        System.out.println("元素" + "b" + "入队");
        myQueue.put("c");
        System.out.println("元素" + "c" + "入队");
        myQueue.put("d");
        System.out.println("元素" + "d" + "入队");
        myQueue.put("e");
        System.out.println("元素" + "e" + "入队");

        System.out.println("此刻队列中元素个数:"+myQueue.getCount());
        System.out.println("队列最多能放元素数量: "+myQueue.getMaxSize());
        System.out.println("队列最少元素数量: "+myQueue.getMinSize());

        // 创建一个放数据的线程
        Thread putThread = new Thread(() -> {
            myQueue.put("f");
            System.out.println("元素" + "f" + "入队");
            myQueue.put("g");
            System.out.println("元素" + "g" + "入队");
        },"put-Thread");

        // 创建一个取数据的线程
        Thread takeThread = new Thread(() -> {
            Object o = myQueue.take();
            System.out.println("元素"+o+"出队");
            o = myQueue.take();
            System.out.println("元素"+o+"出队");
        },"take-Thread");

        // 启动放数据的线程,因为阻塞队列满了,所以放数据线程放不进去f,g,进入wait队列的阻塞状态等待被LOCK.notify唤醒
        System.out.println("-----------put-Thread------------");
        putThread.start();

        // main线程进入普通的阻塞状态2秒,使得放数据的线程putThread一定被执行
        TimeUnit.SECONDS.sleep(2);

        // 启动取数据的线程,取出一个数据后会唤醒进入wait队列的putThread,使其进入锁池的阻塞状态,与其他线程共同竞争cpu
        System.out.println("-----------take-Thread------------");
        takeThread.start();
    }
}

运行结果:如图所示,阻塞队列可用。
分析:
总共有3个线程,main、put-Thread、take-Thread,main线程创建一个最多容纳5个元素的阻塞队列,向阻塞队列依次入队5个元素。put-Thread入队2个元素。take-Thread出队2个元素。对于每一个具体的线程来说,执行顺序没有问题。
对于出队,入队线程的通信来说,调用入队线程的start并没有入队信息打印说明入队线程阻塞了。之后出队线程将队首元素出队,唤醒入队线程,入队线程蒋f入队。之后又进入wait状态。只能进行出队操作。这与脑海中的分析相符。证明完成了阻塞队列的封装。
并发编程核心——九、阻塞队列
备注:

System.out.println(“元素” + “g” + “入队”);在入队操作里面输出,会出现打印前后不一致现象:
System.out.println(“元素” + “f” + “入队”);
System.out.println(“元素” + “a” + “出队”);
所以干脆写到外面了。封装的阻塞队列里面只完成核心的入队,出队,数量增加,减少操作,对于给自己看的显示输出信息解释信息放在外面。

另外就是对于队列属性的获取,使用方法封装了。在封装的类中可以访问自己的私有变量,如果是在其他类中调用必须通过public的获取方法获取变量的值。如果直接使用public修饰变量,不知道情况的人可以直接设置值,队列最小数量变成-1就不好了!!如果被其他人设置成-1,那么就报错或者将其设置为0.

BlockingQueue 即阻塞队列,它是基于 ReentrantLock,依据它的基本原理,我们可 以实现 Web 中的长连接聊天功能,当然其最常用的还是用于实现生产者与消费者模式,大 致如下图所示:
并发编程核心——九、阻塞队列
在 Java 中,BlockingQueue 是一个接口,它的实现类有 ArrayBlockingQueue、 DelayQueue 、 LinkedBlockingDeque 、 LinkedBlockingQueue 、 PriorityBlockingQueue、SynchronousQueue 等,它们的区别主要体现在存储结构上或 对元素操作上的不同,但是对于 take 与 put 操作的原理,却是类似的。

put(an Object):把 an Object 加到 BlockingQueue 里,如果 BlockQueue 没有 空间,则调用此方法的线程被阻断,直到 BlockingQueue 里面有空间再继续。

take:取走 BlockingQueue 里排在首位的对象,若 BlockingQueue 为空,阻断进入 等待状态直到 BlockingQueue 有新的数据被加入


程序员灯塔
转载请注明原文链接:并发编程核心——九、阻塞队列
喜欢 (0)