浅谈Java队列Queue

队列Queue就是一个先进先出的数据结构，与List、Set同一级别，继承了Collection接口。

1、Queue的实现

一、阻塞队列（BlockingQueue）

① 插入：队列不满时可执行插入元素线程，直到队列满。

② 移除：队列不为空均可移除，直到队列为空。

抛出异常：满的时候插入，空的时候取出都会抛异常。

返回特殊值：插入成功返回true

一直阻塞：满时put和空时take会阻塞线程，直到队列可用。

超时退出：满时会阻塞插入一段时间，若是超过必定时间，线程就会退出。

JDK7提供了7个阻塞队列：

① ArrayListBlockingQueue

ArrayBlockingQueue是一个由数组支持的有界阻塞队列。在读写操做上都须要锁住整个容器，所以吞吐量与通常的实现是类似的，适合于实现“生产者消费者”模式。 
② LinkedBlockingQueue

一个由链表结构组成的双向阻塞队列。

基于链表的阻塞队列，内部维持这一关数据缓冲队列。当生产者往队列中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者当即返回；当队列缓冲区达到最大缓冲容量时，才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生成者线程才会被唤醒，反之对消费者的处理也基于一样的原理。

LinkedBlockingQueue之因此可以高效的处理并发数据，还由于其对于生产者和消费者分别采用了独立的锁来控制数据同步，也就意味着在高并发的状况下生产者和消费者能够并行的操做队列中的数据，以此来提升整个队列的并发性能。

③ SynchronousQueue

一个不存储元素的阻塞队列，在某次添加元素后必须等待其余线程取走后才能继续添加。

④ PriorityBlockingQueue 

是一个带优先级的队列，而不是先进先出队列，该队列也没有上限，可是若是队列为空，那么取元素的操做take就会阻塞。

⑤ DelayQueue

是一个存放Delayed 元素的无界阻塞队列，只有在延迟期满时才能从中提取元素。

ArrayListBlockingQueue和LinkedBlockingQueue的区别？

① 队列中锁的实现不一样

ArrayBlockingQueue生产者消费者使用同一个锁。

LinkedBlockingQueue生产用的是putLock，消费是takeLock。

② 在生产和消费时操做不一样

ArrayBlockingQueue实现的队列中在生产和消费的时候，是直接将枚举对象插入或移除的；

LinkedBlockingQueue实现的队列中在生产和消费的时候，须要把枚举对象转换为节点进行插入或移除，会影响性能。

③ 队列大小初始化方式不一样

ArrayBlockingQueue实现的队列必须指定大小

LinkedBlockingQueue能够不指定大小，默认是Integer.MAX_VALUE

ArrayBlockingQueue性能要比LinkedBlockingQueue性能要好，执行速度更快，ArrayBlockingQueue优先使用！

二、非阻塞队列

ConcurrentLinkedQueue是一个基于连接节点的无界线程安全队列，它采用先进先出的规则对节点进行排序，当咱们添加一个元素的时候，它会添加到队列的尾部；当咱们获取一个元素时，它会返回队列头部的元素。

入队和出队操做均利用CAS（compare and set）更新，这样容许多个线程并发执行，而且不会由于加锁而阻塞线程，使得并发性能更好。

注：CAS用于实现多线程同步的原子指令，它将内存位置的内容与给定值进行比较，只有在相同的状况下，将该内存位置的内容修改成新的给定值。 

2、代码实例

package OSChina.Client;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class BlockingQueueTest {
    static final SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
    //定义装苹果的篮子
    public static class Basket{
        // 篮子，可以容纳3个苹果
       static BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);
        //生产苹果，放入篮子
        public void produce() throws InterruptedException{
            queue.put("An apple");
        }
        // 消费苹果，从篮子中取走
        public String consume() throws InterruptedException{
            String apple = queue.take();
            return apple;
        }
        public static int getAppleNumber(){
            return queue.size();
        }
    }

    public static void testBasket(){
        // 创建一个装苹果的篮子
        final Basket basket = new Basket();
        //定义苹果生产者
        class Producer implements Runnable{
            @Override
            public void run() {
                try {
                    while (true){
                        System.out.println("生产者准备生产苹果："+sdf.format(new Date())+"，篮子中苹果数量："+Basket.getAppleNumber());
                        basket.produce();
                        System.out.println("生产者生产苹果完毕：" + sdf.format(new Date())+"，篮子中苹果数量："+Basket.getAppleNumber());
                        Thread.sleep(300);
                    }
                }catch (InterruptedException  ex){
                }
            }
        }
        //定义苹果消费者
        class Consumer implements Runnable{
            @Override
            public void run() {
                try {
                    while (true){
                        System.out.println("消费者准备消费苹果：" + sdf.format(new Date())+"，篮子中苹果数量："+Basket.getAppleNumber());
                        basket.consume();
                        System.out.println("消费完后有苹果：" + sdf.format(new Date())+"，篮子中苹果数量："+Basket.getAppleNumber());
                        Thread.sleep(1000);
                    }
                }catch (InterruptedException ex){
                }
            }
        }

        ExecutorService service = Executors.newFixedThreadPool(2);
        Producer producer = new Producer();
        Consumer consumer = new Consumer();
        service.submit(producer);
        service.submit(consumer);
        // 程序运行10s后，全部任务中止
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        service.shutdownNow();
    }

    public static void main(String[] args) {
        BlockingQueueTest.testBasket();
    }
}

 

江疏影讲Java@目录