java并发库中的阻塞队列--BlockingQueue

时间 2019-11-10

标签 java 并发阻塞队列 blockingqueue 栏目 Java 繁體版

原文原文链接

1.阻塞队列的概念java

阻塞队列与普通队列的区别在于，当队列是空的时，从队列中获取元素的操做将会被阻塞，或者当队列是满时，往队列里添加元素的操做会被阻塞。试图从空的阻塞队列中获取元素的线程将会被阻塞，直到其余的线程往空的队列插入新的元素。一样，试图往已满的阻塞队列中添加新元素的线程一样也会被阻塞，直到其余的线程使队列从新变得空闲起来，如从队列中移除一个或者多个元素，或者彻底清空队列，下图展现了如何经过阻塞队列来合做：编程

线程1往阻塞队列中添加元素，而线程2从阻塞队列中移除元素并发

从刚才的描述能够看出，发生阻塞起码得知足下面至少一个条件： (前提：队列是有界的)socket

1.从队列里取元素时，若是队列为空，则代码一直等在这里（即阻塞），直到队列里有东西了，拿到元素了，后面的代码才能继续ide

2.向队列里放元素时，若是队列满了（即放不下更多元素），则代码也会卡住，直到队列里的东西被取走了（即：有空位能够放新元素了），后面的代码才能继续this

2.生产者消费者模型用阻塞队列实现和原来的区别线程

下面先使用Object.wait()和Object.notify()、非阻塞队列实现生产者-消费者模式：code

public class Test {
    private int queueSize = 10;
    private PriorityQueue<Integer> queue = new PriorityQueue<Integer>(queueSize);
     
    public static void main(String[] args)  {
        Test test = new Test();
        Producer producer = test.new Producer();
        Consumer consumer = test.new Consumer();
         
        producer.start();
        consumer.start();
    }
     
    class Consumer extends Thread{
         
        @Override
        public void run() {
            consume();
        }
         
        private void consume() {
            while(true){
                synchronized (queue) {
                    while(queue.size() == 0){
                        try {
                            System.out.println("队列空，等待数据");
                            queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            queue.notify();
                        }
                    }
                    queue.poll();          //每次移走队首元素
                    queue.notify();
                    System.out.println("从队列取走一个元素，队列剩余"+queue.size()+"个元素");
                }
            }
        }
    }
     
    class Producer extends Thread{
         
        @Override
        public void run() {
            produce();
        }
         
        private void produce() {
            while(true){
                synchronized (queue) {
                    while(queue.size() == queueSize){
                        try {
                            System.out.println("队列满，等待有空余空间");
                            queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            queue.notify();
                        }
                    }
                    queue.offer(1);        //每次插入一个元素
                    queue.notify();
                    System.out.println("向队列取中插入一个元素，队列剩余空间："+(queueSize-queue.size()));
                }
            }
        }
    }
}

这个是经典的生产者-消费者模式，经过阻塞队列和Object.wait()和Object.notify()实现，wait()和notify()主要用来实现线程间通讯。接口

　　具体的线程间通讯方式（wait和notify的使用）在后续问章中会讲述到。队列

　　下面是使用阻塞队列实现的生产者-消费者模式：

public class Test {
    private int queueSize = 10;
    private ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(queueSize);
     
    public static void main(String[] args)  {
        Test test = new Test();
        Producer producer = test.new Producer();
        Consumer consumer = test.new Consumer();
         
        producer.start();
        consumer.start();
    }
     
    class Consumer extends Thread{
         
        @Override
        public void run() {
            consume();
        }
         
        private void consume() {
            while(true){
                try {
                    queue.take();
                    System.out.println("从队列取走一个元素，队列剩余"+queue.size()+"个元素");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
     
    class Producer extends Thread{
         
        @Override
        public void run() {
            produce();
        }
         
        private void produce() {
            while(true){
                try {
                    queue.put(1);
                    System.out.println("向队列取中插入一个元素，队列剩余空间："+(queueSize-queue.size()));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

有没有发现，使用阻塞队列代码要简单得多，不须要再单独考虑同步和线程间通讯的问题。

　　在并发编程中，通常推荐使用阻塞队列，这样实现能够尽可能地避免程序出现意外的错误。

　　阻塞队列使用最经典的场景就是socket客户端数据的读取和解析，读取数据的线程不断将数据放入队列，而后解析线程不断从队列取数据解析。还有其余相似的场景，只要符合生产者-消费者模型的均可以使用阻塞队列。

3.实现原理：

这里只贴几段主要的代码，体会一下思想：

/** Main lock guarding all access */
final ReentrantLock lock;
 
/** Condition for waiting takes */
private final Condition notEmpty;
 
/** Condition for waiting puts */
private final Condition notFull;

这3个变量很重要，ReentrantLock重入锁，notEmpty检查不为空的Condition 以及 notFull用来检查队列未满的Condition

Condition是一个接口，里面有二个重要的方法：
await() ： Causes the current thread to wait until it is signalled or interrupted. 即阻塞当前线程，直到被通知（唤醒）或中断

singal(): Wakes up one waiting thread. 唤醒阻塞的线程

再来看put方法：(jdk 1.8)

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length)
            notFull.await();
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

1.先获取锁

2.而后用while循环检测元素个数是否等于items长度，若是相等，表示队列满了,调用notFull的await()方法阻塞线程

3.不然调用enqueue()方法添加元素

4.最后解锁

private void enqueue(E x) {
    // assert lock.getHoldCount() == 1;
    // assert items[putIndex] == null;
    final Object[] items = this.items;
    items[putIndex] = x;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
    notEmpty.signal();
}

这是添加元素的代码（jdk 1.8）,注意最后一行notEmpty.signal()方法，表示添加完元素后，调用singal()通知等待（从队列中取元素）的线程，队列不空(有值)啦，能够来取东西了。

相似的take()与dequeue()方法则至关于逆过程（注：一样都是jdk 1.8）

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}

相似的：

1. 先加锁

2. 若是元素个数为空，表示队列已空，调用notEmpty的await()阻塞线程，直接队列里又有新元素加入为止

3. 而后调用dequeue 从队列里删除元素

4. 解锁

dequeue方法：

private E dequeue() {
    // assert lock.getHoldCount() == 1;
    // assert items[takeIndex] != null;
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    notFull.signal();
    return x;
}

倒数第2行，元素移除后，调用notFull.singnal唤醒等待（向队列添加元素的）线程，队列有空位了，能够向里面添加元素了。