用法

Queue也就是队列，只能有两种基本的操作，在头部取走一个元素和在尾部增加一个元素，所以是一种FIFO结构（先进先出），不同于栈，栈是一种后进先出的数据结构。

阻塞Queue常用的方法：

add        增加一个元索                     如果队列已满，则抛出一个IIIegaISlabEepeplian异常
remove   移除并返回队列头部的元素    如果队列为空，则抛出一个NoSuchElementException异常
element  返回队列头部的元素             如果队列为空，则抛出一个NoSuchElementException异常
offer       添加一个元素并返回true       如果队列已满，则返回false
poll         移除并返问队列头部的元素    如果队列为空，则返回null
peek       返回队列头部的元素             如果队列为空，则返回null
put         添加一个元素                      如果队列满，则阻塞
take        移除并返回队列头部的元素     如果队列为空，则阻塞

其中poll和take都是取走队列头部的元素，区别在于take是阻塞的，如果take的时候，队列为空，那么调用take方法的线程会一直阻塞知道队列元素不为空就会被唤醒，而poll不是阻塞的，poll方法只会检查此刻队列的状态，若为空则返回null。

同理put和offer也是同一个道理。
应用

阻塞队列对应的是生产者消费者模型，在线程池中有所应用，线程池的内部就是一个阻塞队列，生产者往里面增加任务，而消费者也就是线程池中的线程就会不断地从阻塞队列中取出任务然后消费。

BlockQueue有两种一种是基于数组这种数据结构的ArrayBlockQueue，另外一种是基于链表LinkedBlockQueue。



源码

首先BlockQueue是线程安全的，其内部有一个ReentrantLock，由ReentrantLock产生两个Condition，其中一个Condition是调用take方法阻塞的线程集合，另外一个是调用put方法阻塞线程的集合，我们知道ReentrantLock是基于AQS，那么Condition就是对这些阻塞线程操作的封装，主要用于阻塞线程和唤醒在队列中等待线程。



take方法

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


首先lock加锁，然后循环检查队列中元素的个数，若为0则阻塞，若不为0，则返回一个元素并删除，dequeue的具体细节和用不同的数据结构实现是不一样的，BlcokQueue一般有基于数组数据结构实现，也有基于链表数据结构实现的。

其他的方法看了下都累死，都是检查状态，要不阻塞，要不执行成功。



一个对比

阻塞队列是线程安全的，那么我们和LinkedList来做一个对比，LinkedList不是线程安全的，程序就是几个不同的线程不断的在尾部增加1000个元素：

public class BlockQueueTest {

    private LinkedBlockingQueue<Integer> collection;

    private CountDownLatch countDownLatch;

    public BlockQueueTest(CountDownLatch countDownLatch) {
        this.countDownLatch = countDownLatch;
        collection = new LinkedBlockingQueue<Integer>();
    }

    private void addElement(final int ele) {
        for (int i = 0 ; i < 10; i ++) {
            new Thread() {
                @Override
                public void run() {
                    for (int j = 0; j < 1000000; j ++)
                        collection.add(ele);

                    System.out.println(Thread.currentThread().getName() + " finish");
                    countDownLatch.countDown();
                }
            }.start();
        }
    }

    public LinkedBlockingQueue<Integer> getCollection() {
        return collection;
    }

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(10);
        BlockQueueTest test = new BlockQueueTest(countDownLatch);
        test.addElement(10);
        countDownLatch.await();
        System.out.println(test.getCollection().size());
    }


}


运行结果是：

Thread-5 finish
Thread-4 finish
Thread-2 finish
Thread-3 finish
Thread-7 finish
Thread-6 finish
Thread-9 finish
Thread-0 finish
Thread-1 finish
Thread-8 finish
10000000


若使用LinkedList来测试：

public class BlockQueueTest {

    private LinkedList<Integer> collection;

    private CountDownLatch countDownLatch;

    public BlockQueueTest(CountDownLatch countDownLatch) {
        this.countDownLatch = countDownLatch;
        collection = new LinkedList<Integer>();
    }

    private void addElement(final int ele) {
        for (int i = 0 ; i < 10; i ++) {
            new Thread() {
                @Override
                public void run() {
                    for (int j = 0; j < 1000000; j ++)
                        collection.add(ele);

                    System.out.println(Thread.currentThread().getName() + " finish");
                    countDownLatch.countDown();
                }
            }.start();
        }
    }

    public LinkedList<Integer> getCollection() {
        return collection;
    }

    public static void main(String[] args) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(10);
        BlockQueueTest test = new BlockQueueTest(countDownLatch);
        test.addElement(10);
        countDownLatch.await();
        System.out.println(test.getCollection().size());
    }


}


测试结果：

Thread-5 finish
Thread-4 finish
Thread-1 finish
Thread-2 finish
Thread-0 finish
Thread-6 finish
Thread-9 finish
Thread-3 finish
Thread-8 finish
Thread-7 finish
1294411


由于不是线程安全的，最终集合中的元素不是我们预期的10000000

		

			BlockQueue 解析
		
		

		

			
生产者、消费者模式
https://www.jianshu.com/p/024a36b83099












		
		
posted @ 2019-04-02 16:23 诸葛子房 阅读(...) 评论(...)  编辑 收藏
	

基本原理

特殊的队列:BlockingQueue

如果BlockQueue是空的,从BlockingQueue取东西的操作将会被阻断进入等待状态,直到BlockingQueue进了东西才会被唤醒.

同样,如果BlockingQueue是满的,任何试图往里存东西的操作也会被阻断进入等待状态,直到BlockingQueue里有空间才会被唤醒继续操作. 

使用BlockingQueue的关键技术点如下: 



BlockingQueue的常用方法



1)add(anObject) 

  把anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则报异常

2)offer(anObject) 

  表示如果可能的话,将anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则返回false.

3)put(anObject)

4)poll(time) 

  取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回null。

5)take() 

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

BlockingQueue四个具体的实现类

1)ArrayBlockingQueue 

规定大小的BlockingQueue,其构造函数必须带一个int参数来指明其大小.其所含的对象是以FIFO(先入先出)顺序排序的. 

2)LinkedBlockingQueue 

大小不定的BlockingQueue,若其构造函数带一个规定大小的参数,生成的BlockingQueue有大小限制,若不带大小参数,所生成的BlockingQueue的大小由Integer.MAX_VALUE来决定.其所含的对象是以FIFO(先入先出)顺序排序的 

3)PriorityBlockingQueue 

类似于LinkedBlockQueue,但其所含对象的排序不是FIFO,而是依据对象的自然排序顺序或者是构造函数的Comparator决定的顺序. 

4)SynchronousQueue 

特殊的BlockingQueue,对其的操作必须是放和取交替完成的. 

LinkedBlockingQueue和ArrayBlockingQueue比较起来,它们背后所用的数据结构不一样,导致LinkedBlockingQueue的数据吞吐量要大于ArrayBlockingQueue,但在线程数量很大时其性能的可预见性低于ArrayBlockingQueue.  

package com.example.demo.webservice;

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(int size) {
        this.index = 0;
        this.removeIndex = 0;
        this.currLen = 0;
        this.size = size;
        queue = new Object[size];
    }

    public BlockQueue() {
        this(10);
    }
    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);
        Thread t1=new Thread(()->{
            for(int i=0;i<100;i++){
                try {
                    System.out.println("生产者生产的数字是： "+i);
                    blockQueue.push(i);
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        Thread t2=new Thread(()->{
            while(true){
                try {
                    System.out.println("消费者消费的的数字是： "+blockQueue.pop());
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });

        t1.start();
        t2.start();

    }

}