并发编程(7)-阻塞队列

电玩女神 2022-06-06 09:19 418阅读 0赞

BlockingQueue：顾名思义，首先它是一个队列，并且支持阻塞的机制，阻塞的放入和得到数据。我们要实现LinkedBlockingQueue下面两个简单的方法put和take。  
put(anObject)：把anObject加入到BlockingQueue里面，如果BlockingQueue没有空间，则调用此方法的线程被阻断，直到BlockingQueue里面有空间再继续。  
take：取走BlockingQueue里面排在首位的对象，若BlockingQueue为空，阻断进入等待状态直到BlockingQueue有新的数据被加入。

![这里写图片描述][SouthEast]

自定义一个阻塞队列小例子

package com.thread.mythread.conn009;
    
    import java.util.LinkedList;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class MyQueue {
    
        //队列容器，存放放入的元素
        private final LinkedList<Object> list = new LinkedList<Object>();
        //容量计数器，记录每次放入容器中元素的数量
        private final AtomicInteger count = new AtomicInteger(0);
        //队列容量值
        private final int maxSize;
        private final int minSize = 0;
    
        private final Object lock = new Object();
    
        //初始化是设置队列的最大容量
        public MyQueue (int maxSize){
            this.maxSize = maxSize;
        }
    
        public void put (Object obj) {
            synchronized(lock){
                //当队列容器容量已经达到最大值，进入阻塞状态
                while(count.get() == maxSize){
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                //当队列容量未达到最大值，向容器中添加元素
                list.add(obj);
                //记录队列容器中的元素数量
                count.getAndIncrement();
                System.out.println(" 元素 " + obj + " 被添加 ");
                //通知take方法可以从队列容器中取走元素
                lock.notify();
    
            }
        }
    
        public Object take(){
            Object temp = null;
            synchronized (lock) {
                //如果队列容器中元素个数为零，进入阻塞状态
                while(count.get() == minSize){
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                //减少队列容器中元素的个数
                count.getAndDecrement();
                //移除队列容器中的第一个元素
                temp = list.removeFirst();
                System.out.println(" 元素 " + temp + " 被消费 ");
                //通知put方法，又有剩余空间可以放入新的元素
                lock.notify();
            }
            return temp;
        }
    
        public int size(){
            return count.get();
        }
    
        public static void main(String[] args) throws Exception {
    
            final MyQueue m = new MyQueue(5);
            m.put("a");
            m.put("b");
            m.put("c");
            m.put("d");
            m.put("e");
            System.out.println("当前元素个数：" + m.size());
            Thread t1 = new Thread(new Runnable() {
                public void run() {
                    m.put("h");
                    m.put("i");
                }
            }, "t1");
    
            Thread t2 = new Thread(new Runnable() {
                public void run() {
                    try {
                        Thread.sleep(1000);
                        Object t1 = m.take();
                        System.out.println("被取走的元素为：" + t1);
                        Thread.sleep(1000);
                        Object t2 = m.take();
                        System.out.println("被取走的元素为：" + t2);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, "t2");
    
            t1.start();
            Thread.sleep(1000);
            t2.start();
        }
    }

打印结果：

> 元素 a 被添加  
> 元素 b 被添加  
> 元素 c 被添加  
> 元素 d 被添加  
> 元素 e 被添加  
> 当前元素个数：5  
> 元素 a 被消费  
> 元素 h 被添加  
> 被取走的元素为：a  
> 元素 b 被消费  
> 元素 i 被添加  
> 被取走的元素为：b

[SouthEast]: /images/20220606/78611511cecc4db1b2ec916b068750eb.png