• BlockingQueue  阻塞算法
• ConcurrentLinkedQueue，非阻塞算法

• add(e) remove() element() 方法不会阻塞线程。当不满足约束条件时，会抛出IllegalStateException 异常。例如：当队列被元素填满后，再调用add(e)，则会抛出异常。
• offer(e) poll() peek() 方法即不会阻塞线程，也不会抛出异常。例如：当队列被元素填满后，再调用offer(e)，则不会插入元素，函数返回false。
• 要想要实现阻塞功能，需要调用put(e) take() 方法。当不满足约束条件时，会阻塞线程。

Java代码   

1. public boolean add(E e) {
2. if (offer(e))
3. return true;
4. else
5. throw new IllegalStateException("Queue full");//队列已满，抛异常
6. }
7. 
   
8. public E remove() {
9. E x = poll();
10. if (x != null)
11. return x;
12. else
13. throw new NoSuchElementException();//队列为空，抛异常
14. }
    

Java代码   

1. public boolean offer(E e) {
2. if (e == null)throw new NullPointerException();
3. final ReentrantLock lock = this.lock;
4. lock.lock();
5. try {
6. if (count == items.length)//队列已满，返回false
7. return false;
8. else {
9. insert(e);//insert方法中发出了notEmpty.signal();
10. return true;
11. }
12. } finally {
13. lock.unlock();
14. }
15. }
16. 
    
17. public E poll() {
18. final ReentrantLock lock = this.lock;
19. lock.lock();
20. try {
21. if (count == 0)//队列为空，返回false
22. return null;
23. E x = extract();//extract方法中发出了notFull.signal();
24. return x;
25. } finally {
26. lock.unlock();
27. }
28. }
    

Java代码   

1. public void put(E e)throws InterruptedException {
2. if (e == null)throw new NullPointerException();
3. final E[] items = this.items;
4. final ReentrantLock lock = this.lock;
5. lock.lockInterruptibly();
6. try {
7. try {
8. while (count == items.length)//如果队列已满，等待notFull这个条件，这时当前线程被阻塞
9. notFull.await();
10. } catch (InterruptedException ie) {
11. notFull.signal(); //唤醒受notFull阻塞的当前线程
12. throw ie;
13. }
14. insert(e);
15. } finally {
16. lock.unlock();
17. }
18. }
19. 
    
20. public E take() throws InterruptedException {
21. final ReentrantLock lock = this.lock;
22. lock.lockInterruptibly();
23. try {
24. try {
25. while (count == 0)//如果队列为空，等待notEmpty这个条件，这时当前线程被阻塞
26. notEmpty.await();
27. } catch (InterruptedException ie) {
28. notEmpty.signal();//唤醒受notEmpty阻塞的当前线程
29. throw ie;
30. }
31. E x = extract();
32. return x;
33. } finally {
34. lock.unlock();
35. }
36. }
37.  

• ArrayBlockingQueue，其构造函数必须带一个int参数来指明其大小
• LinkedBlockingQueue，若其构造函数带一个规定大小的参数，生成的BlockingQueue有大小限制，若不带大小参数，所生成的BlockingQueue的大小由Integer.MAX_VALUE来决定
• PriorityBlockingQueue，其所含对象的排序不是FIFO,而是依据对象的自然排序顺序或者是构造函数的Comparator决定的顺序

Java代码   

1. static class Node<E> {
2. /** The item, volatile to ensure barrier separating write and read */
3. volatile E item;
4. Node<E> next;
5. Node(E x) { item = x; }
6. }
   

Java代码   

1. /** 头指针 */
2. private transient Node<E> head;//head.next是队列的头元素
3. /** 尾指针 */
4. private transient Node<E> last;//last.next是null
   

Java代码   

1. private void enqueue(E x) {
2. last = last.next = new Node<E>(x);//入队是为last再找个下家
3. }
4. 
   
5. private E dequeue() {
6. Node<E> first = head.next; //出队是把head.next取出来，然后将head向后移一位
7. head = first;
8. E x = first.item;
9. first.item = null;
10. return x;
11. }
    

Java代码   

1. private final AtomicInteger count =new AtomicInteger(0);
2. /** 用于读取的独占锁*/
3. private final ReentrantLock takeLock =new ReentrantLock();
4. /** 队列是否为空的条件 */
5. private final Condition notEmpty = takeLock.newCondition();
6. /** 用于写入的独占锁 */
7. private final ReentrantLock putLock =new ReentrantLock();
8. /** 队列是否已满的条件 */
9. private final Condition notFull = putLock.newCondition();
   

Java代码   

1. public boolean offer(E e) {
2. if (e == null)throw new NullPointerException();
3. final AtomicInteger count = this.count;
4. if (count.get() == capacity)
5. return false;
6. int c = -1;
7. final ReentrantLock putLock =this.putLock;//入队当然用putLock
8. putLock.lock();
9. try {
10. if (count.get() < capacity) {
11. enqueue(e); //入队
12. c = count.getAndIncrement(); //队长度+1
13. if (c + 1 < capacity)
14. notFull.signal(); //队列没满，当然可以解锁了
15. }
16. } finally {
17. putLock.unlock();
18. }
19. if (c == 0)
20. signalNotEmpty();//这个方法里发出了notEmpty.signal();
21. return c >= 0;
22. }
23. 
    
24. public E poll() {
25. final AtomicInteger count = this.count;
26. if (count.get() == 0)
27. return null;
28. E x = null;
29. int c = -1;
30. final ReentrantLock takeLock =this.takeLock;出队当然用takeLock
31. takeLock.lock();
32. try {
33. if (count.get() > 0) {
34. x = dequeue();//出队
35. c = count.getAndDecrement();//队长度-1
36. if (c > 1)
37. notEmpty.signal();//队列没空，解锁
38. }
39. } finally {
40. takeLock.unlock();
41. }
42. if (c == capacity)
43. signalNotFull();//这个方法里发出了notFull.signal();
44. return x;
45. }
    

Java代码   

1. public boolean offer(E e) {
2. if (e == null)throw new NullPointerException();
3. Node<E> n = new Node<E>(e, null);
4. for (;;) {
5. Node<E> t = tail;
6. Node<E> s = t.getNext();
7. if (t == tail) { //------------------------------a
8. if (s == null) {//---------------------------b
9. if (t.casNext(s, n)) { //-------------------c
10. casTail(t, n); //------------------------d
11. return true;
12. }
13. } else {
14. casTail(t, s); //----------------------------e
15. }
16. }
17. }
18. }
    

Java代码   

1. if(!queue.isEmpty()) {
2. queue.poll(obj);
3. }
   

Java代码   

1. synchronized(queue) {
2. if(!queue.isEmpty()) {
3. queue.poll(obj);
4. }
5. }
   

Java代码   

1. public finalclass Counter {
2. private long value =http://www.mamicode.com/0;
3. public synchronizedlong getValue() {
4. return value;
5. }
6. public synchronizedlong increment() {
7. return ++value;
8. }
9. }
   

Java代码   

1. public class NonblockingCounter {
2. private AtomicInteger value;//前面提到过，AtomicInteger类是以原子的方式操作整型变量。
3. public int getValue() {
4. return value.get();
5. }
6. public int increment() {
7. int v;
8. do {
9. v = value.get();
10. while (!value.compareAndSet(v, v +1));
11. return v + 1;
12. }
13. }
    

Java代码   

1. public class LinkedQueue <E> {
2. private staticclass Node <E> {
3. final E item;
4. final AtomicReference<Node<E>> next;
5. Node(E item, Node<E> next) {
6. this.item = item;
7. this.next = new AtomicReference<Node<E>>(next);
8. }
9. }
10. private AtomicReference<Node<E>> head
11. = new AtomicReference<Node<E>>(new Node<E>(null,null));
12. private AtomicReference<Node<E>> tail = head;
13. public boolean put(E item) {
14. Node<E> newNode = new Node<E>(item,null);
15. while (true) {
16. Node<E> curTail = tail.get();
17. Node<E> residue = curTail.next.get();
18. if (curTail == tail.get()) {
19. if (residue == null)/* A */ {
20. if (curTail.next.compareAndSet(null, newNode))/* C */ {
21. tail.compareAndSet(curTail, newNode) /* D */ ;
22. return true;
23. }
24. } else {
25. tail.compareAndSet(curTail, residue) /* B */;
26. }
27. }
28. }
29. }
30. }
    

上图显示的是：有两个元素，处在静止状态的队列 

插入操作在插入新元素（A）之前，先检查队列是否处在中间状态。如果是在中间状态，那么肯定有其他线程已经处在元素插入的中途，在步骤（C）和（D）之间。不必等候其他线程完成，当前线程就可以 “帮助” 它完成操作，把尾指针向前移动（B）。如果有必要，它还会继续检查尾指针并向前移动指针，直到队列处于静止状态，这时它就可以开始自己的插入了。
第一个 CAS（C）可能因为两个线程竞争访问队列当前的最后一个元素而失败；在这种情况下，没有发生修改，失去 CAS 的线程会重新装入尾指针并再次尝试。如果第二个 CAS（D）失败，插入线程不需要重试 —— 因为其他线程已经在步骤（B）中替它完成了这个操作！

 
上图显示的是：处在插入中间状态的队列，在新元素插入之后，尾指针更新之前 

 
上图显示的是：在尾指针更新后，队列重新处在静止状态

Java多线程之线程安全队列Queue