BlockingQueue介绍

几种类型的BlockingQueue

ArrayBlockingQueue ：一个由数组结构组成的有界阻塞队列。

LinkedBlockingQueue ：一个由链表结构组成的有界阻塞队列。

PriorityBlockingQueue ：一个支持优先级排序的无界阻塞队列。

DelayQueue：一个使用优先级队列实现的无界阻塞队列。

SynchronousQueue：一个不存储元素的阻塞队列。

LinkedTransferQueue：一个由链表结构组成的无界阻塞队列。

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

ArrayBlockingQueue是一个用数组实现的有界阻塞队列。此队列按照先进先出（FIFO）的原则对元素进行排序。默认情况下不保证访问者公平的访问队列，所谓公平访问队列是指阻塞的所有生产者线程或消费者线程，当队列可用时，可以按照阻塞的先后顺序访问队列，即先阻塞的生产者线程，可以先往队列里插入元素，先阻塞的消费者线程，可以先从队列里获取元素。通常情况下为了保证公平性会降低吞吐量。

LinkedBlockingQueue是一个用链表实现的有界阻塞队列。此队列的默认和最大长度为Integer.MAX_VALUE。此队列按照先进先出的原则对元素进行排序。

PriorityBlockingQueue是一个支持优先级的无界队列。默认情况下元素采取自然顺序排列，也可以通过比较器comparator来指定元素的排序规则。元素按照升序排列。

DelayQueue是一个支持延时获取元素的无界阻塞队列。队列使用PriorityQueue来实现。队列中的元素必须实现Delayed接口，在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素。我们可以将DelayQueue运用在以下应用场景：

• 缓存系统的设计：可以用DelayQueue保存缓存元素的有效期，使用一个线程循环查询DelayQueue，一旦能从DelayQueue中获取元素时，表示缓存有效期到了。 
• 定时任务调度。使用DelayQueue保存当天将会执行的任务和执行时间，一旦从DelayQueue中获取到任务就开始执行，从比如TimerQueue就是使用DelayQueue实现的。

关于BlockingQueue的drainTo方法

int drainTo(Collection<? super E> c,  int maxElements)

• 从此队列中移除给定数量的可用元素，并将这些元素添加到给定 collection 中。
• 在向 collection c 中添加元素没有成功时，可能导致在抛出相关异常时，元素会同时在两个 collection 中出现，或者在其中一个 collection 中出现，也可能在两个 collection 中都不出现。
• 如果试图将一个队列放入自身队列中，则会导致 IllegalArgumentException 异常。
• 如果正在进行此操作时，正在修改指定的 collection，则此操作行为是不确定的。  

使用技巧

    //先从queue中获取一个对象，如果没有对象线程自动阻塞

    FixERForFixOutgoingValue firstValue = stkRequestQueue.take();

    list.add(firstValue);

    //如果获取到对象了，则一次性获取剩下的全部对象

    stkRequestQueue.drainTo(list);  

BlockingQueue 构建生产消费者模式：

1package lands.concurrent;
2
3import java.util.concurrent.BlockingQueue;
4
5import java.util.concurrent.LinkedBlockingQueue;
6
7public class BlockingQueueTest {
8
9    /**
10
11     * @param args
12
13     */
14
15    public static void main(String[] args) {
16
17        //BlockingQueue q = new ArrayBlockingQueue(100); //在构造时需要指定容量
18
19        BlockingQueue q = new LinkedBlockingQueue(); //在构造时默认没有上限，但也可以选择指定最大上限
20
21        BlockingQueueProducer p1 = new BlockingQueueProducer(q);
22
23        BlockingQueueProducer p2 = new BlockingQueueProducer(q);
24
25        new Thread(p1).start();
26
27        new Thread(p2).start();
28
29        BlockingQueueConsumer c1 = new BlockingQueueConsumer(q);
30
31        BlockingQueueConsumer c2 = new BlockingQueueConsumer(q);
32
33        new Thread(c1).start();
34
35        new Thread(c2).start();
36
37    }
38
39}
40
41class BlockingQueueProducer implements Runnable {
42
43    private final BlockingQueue queue;
44
45    BlockingQueueProducer(BlockingQueue q) {
46
47        queue = q;
48
49    }
50
51    public void run() {
52
53        try {
54
55            int i = 0;
56
57            while (true) {
58
59                i++;
60
61                queue.put(produce(i));
62
63                //System.out.println("remainingCapacity:" + queue.remainingCapacity());
64
65            }
66
67        } catch (InterruptedException ex) {
68
69            ex.printStackTrace();
70
71        }
72
73    }
74
75    String produce(int i) {
76
77        //create your wanted object
78
79        return i + "";
80
81    }
82
83}
84
85class BlockingQueueConsumer implements Runnable {
86
87    private final BlockingQueue queue;
88
89    BlockingQueueConsumer(BlockingQueue q) {
90
91        queue = q;
92
93    }
94
95    public void run() {
96
97        try {
98
99            while (true) {
100
101                consume(queue.take());
102
103            }
104
105        } catch (InterruptedException ex) {
106
107            ex.printStackTrace();
108
109        }
110
111    }
112
113    void consume(Object x) {
114
115        //use a object in queue
116
117        System.out.println(x);
118
119    }
120
121}