Java实现阻塞队列+生产者消费者

﹏ヽ暗。殇╰゛Y 2021-09-21 05:26 396阅读 0赞

### Java实现阻塞队列 ###

package test.blockqueue;
    
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class BlockQueue<T> { 
    
        private int size;
        private Object[] queue;
    
        private Lock lock = new ReentrantLock();
        private Condition full = lock.newCondition();
        private Condition empty = lock.newCondition();
    
        private int index;
        private int removeIndex;
        private int currLen;
    
        public BlockQueue() { 
            this(10);
        }
    
        public BlockQueue(int size) { 
            this.index = 0;
            this.removeIndex = 0;
            this.currLen = 0;
            this.size = size;
            queue = new Object[size];
        }
    
        public void push(T element) throws InterruptedException { 
            lock.lock();
            try { 
                while (currLen == size) { 
                    System.out.println("队列满。。。");
                    full.await();
                }
                queue[index] = element;
                if (++index == size) { 
                    index = 0;
                }
                currLen++;
                empty.signal();
            } finally { 
                lock.unlock();
            }
        }
    
        public T pop() throws InterruptedException { 
            lock.lock();
            try { 
                while (currLen == 0) { 
                    System.out.println("队列空。。。");
                    empty.await();
                }
                Object obj = queue[removeIndex];
                if (++removeIndex == size) { 
                    removeIndex = 0;
                }
                currLen--;
                full.signal();
                return (T) obj;
            } finally { 
                lock.unlock();
            }
        }
    
        public static void main(String[] args) throws InterruptedException { 
            BlockQueue<Integer> blockQueue = new BlockQueue<Integer>(3);
            blockQueue.push(1);
            // System.out.println(blockQueue.pop());
            blockQueue.push(2);
            // System.out.println(blockQueue.pop());
            blockQueue.push(3);
            // System.out.println(blockQueue.pop());
    
            blockQueue.push(5);
            blockQueue.push(5);
            System.out.println(blockQueue.pop());
        }
    
    }

Condition 是 Java 提供了来实现等待 / 通知的类，Condition 类还提供比 wait/notify 更丰富的功能，Condition 对象是由 lock 对象所创建的。但是同一个锁可以创建多个 Condition 的对象，即创建多个对象监视器。这样的好处就是可以指定唤醒线程。notify 唤醒的线程是随机唤醒一个。

condition 对象通过 lock.newCondition() 来创建，**用 condition.await（）来实现让线程等待，使线程进入阻塞**。**用 condition.signal() 来实现唤醒线程**。唤醒的线程是用同一个 conditon 对象调用 await() 方法而进入阻塞。并且和 wait/notify 一样，await（）和 signal（）也是在同步代码区内执行。

--------------------

### Java中的阻塞队列BlockingQueu ###

*  **ArrayBlockingQueue：BlockingQueue的实现类**
    
    基于数组的阻塞队列实现，在 ArrayBlockingQueue 内部，维护了一个定长数组，以便缓存队列中的数据对象，这是一个常用的阻塞队列，除了一个定长数组外，ArrayBlockingQueue 内部还保存着两个整形变量，分别标识着队列的头部和尾部在数组中的位置。
 *  **LinkedBlockingQueue：BlockingQueue的实现类**
    
    基于链表的阻塞队列，同 ArrayListBlockingQueue 类似，其内部也维持着一个数据缓冲队列（该队列由一个链表构成），当生产者往队列中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者立即返回；只有当队列缓冲区达到最大值缓存容量时（LinkedBlockingQueue 可以通过构造函数指定该值），才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生产者线程会被唤醒，反之对于消费者这端的处理也基于同样的原理。
 *  ArrayBlockingQueue **在生产者放入数据和消费者获取数据，都是共用同一个锁对象**，由此也意味着**两者无法真正并行运行**，这点尤其不同于 LinkedBlockingQueue；LinkedBlockingQueue 之所以能够高效的处理并发数据，还因为其对于生产者端和消费者端分别采用了独立的锁来控制数据同步，这也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。
 *  ArrayBlockingQueue 和 LinkedBlockingQueue 间还有一个明显的不同之处在于，**前者在插入或删除元素时不会产生或销毁任何额外的对象实例，而后者则会生成一个额外的 Node 对象**。这在长时间内需要高效并发地处理大批量数据的系统中，其对于 GC 的影响还是存在一定的区别。而在创建 ArrayBlockingQueue 时，我们还可以控制对象的内部锁是否采用公平锁，默认采用非公平锁。

--------------------

### Java实现生产者消费者 ###

生产者消费者的实现方式有多种，常见的是await/signal 和 阻塞队列

**第一种方式await/signal，实现方式跟阻塞队列十分相似。**

import java.util.LinkedList;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class Storage { 
    
        // 仓库最大存储量
        private final int MAX_SIZE = 10;
        // 仓库存储的载体
        private LinkedList<Object> list = new LinkedList<Object>();
        // 锁
        private final Lock lock = new ReentrantLock();
        // 仓库满的条件变量
        private final Condition full = lock.newCondition();
        // 仓库空的条件变量
        private final Condition empty = lock.newCondition();
    
        public void produce()
        { 
            // 获得锁
            lock.lock();
            while (list.size() + 1 > MAX_SIZE) { 
                System.out.println("【生产者" + Thread.currentThread().getName()
    		             + "】仓库已满");
                try { 
                    full.await();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
            list.add(new Object());
            System.out.println("【生产者" + Thread.currentThread().getName() 
    				 + "】生产一个产品，现库存" + list.size());
    
            empty.signalAll();
            lock.unlock();
        }
    
        public void consume()
        { 
            // 获得锁
            lock.lock();
            while (list.size() == 0) { 
                System.out.println("【消费者" + Thread.currentThread().getName()
    		             + "】仓库为空");
                try { 
                    empty.await();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
            list.remove();
            System.out.println("【消费者" + Thread.currentThread().getName()
    		         + "】消费一个产品，现库存" + list.size());
    
            full.signalAll();
            lock.unlock();
        }
    }

**阻塞队列实现方式：**

import java.util.concurrent.LinkedBlockingQueue;
    
    public class Storage { 
    
        // 仓库存储的载体
        private LinkedBlockingQueue<Object> list = new LinkedBlockingQueue<>(10);
    
        public void produce() { 
            try{ 
                list.put(new Object());
                System.out.println("【生产者" + Thread.currentThread().getName()
                        + "】生产一个产品，现库存" + list.size());
            } catch (InterruptedException e){ 
                e.printStackTrace();
            }
        }
    
        public void consume() { 
            try{ 
                list.take();
                System.out.println("【消费者" + Thread.currentThread().getName()
                        + "】消费了一个产品，现库存" + list.size());
            } catch (InterruptedException e){ 
                e.printStackTrace();
            }
        }
    }

### 补充：同步队列 ###

AQS 中很重要的一个数据结构就是同步队列了，他是一个 FIFO 双向的同步队列。

当前的线程在获取同步状态失败时，同步器会将当前线程以及等待状态等信息构造成一个节点（Node）并将其加入同步队列队尾，同时会阻塞当前线程，当同步状态释放时，会把首节点中的线程唤醒，使其再次尝试获取同步状态。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzOTQ1MjQ2_size_16_color_FFFFFF_t_70]

--------------------

**【Java 面试那点事】**

**这里致力于分享 Java 面试路上的各种知识，无论是技术还是经验，你需要的这里都有！**

**这里可以让你【快速了解 Java 相关知识】，并且【短时间在面试方面有跨越式提升】**

**面试路上，你不孤单！**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzOTQ1MjQ2_size_16_color_FFFFFF_t_70 1]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzOTQ1MjQ2_size_16_color_FFFFFF_t_70]: /images/20210920/d65e51fa9f124c3da6cc161b4248f1bb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzOTQ1MjQ2_size_16_color_FFFFFF_t_70 1]: /images/20210920/6ec55e3f0e3b400f9db9550f93888f2f.png