java集合包LinkedBlockingQueue底层源码分析

　　一、demopublic class LinkedBlockingQueueTest {     public static void main(String[] args) throws Exception{         LinkedBlockingQueue linkedBlockingQueue = new LinkedBlockingQueue(10);         linkedBlockingQueue.put(＂今天头条＂);         System.out.println(linkedBlockingQueue.take());         Iterator iterator = linkedBlockingQueue.iterator();         while (iterator.hasNext()){             System.out.println(iterator.next());         }     } }二、构造方法分析
　　Condition：如果你获取了一把锁，然后调用了Condition.await()，此时会释放锁，当前线程会进入一个cnodition等待队列，事后人家来唤醒你，Cnodition.signal()，这个时候会从condition等待队列中把你加入到wait等待队列里去，然后人家释放锁，就会唤醒wait等待队列里的线程尝试来获取锁；		/** 当前队列元素数量 */	 		private final AtomicInteger count = new AtomicInteger(); 		/** Lock held by take, poll, etc */ 		/** 获取队列元素时加的锁 */     private final ReentrantLock takeLock = new ReentrantLock();      /** Wait queue for waiting takes */     private final Condition notEmpty = takeLock.newCondition();      /** Lock held by put, offer, etc */ 		/** 放入元素时加的锁*/     private final ReentrantLock putLock = new ReentrantLock();      /** Wait queue for waiting puts */     private final Condition notFull = putLock.newCondition();  public LinkedBlockingQueue(int capacity) {     if (capacity <= 0) throw new IllegalArgumentException();   	/** 队列最大大小 */     this.capacity = capacity;   	/** last和head指针指向空节点 */     last = head = new Node(null); }三、put方法public void put(E e) throws InterruptedException {     if (e == null) throw new NullPointerException();     // Note: convention in all put/take/etc is to preset local var     // holding count negative to indicate failure unless set.     int c = -1;   	/** 创建一个单向节点 */     Node node = new Node(e);   	/** 获取添加元素的锁 */     final ReentrantLock putLock = this.putLock;   	/** 获取队列当前数量 */     final AtomicInteger count = this.count;   	/** 加一个可以被中断的锁，保证同一时间只有一个线程可以添加数据     	* 如果加锁线程被中断了，此时加锁会失败，抛出一个线程中断的异常出来       */     putLock.lockInterruptibly();     try {         /*          * Note that count is used in wait guard even though it is          * not protected by lock. This works because count can          * only decrease at this point (all other puts are shut          * out by lock), and we (or some other waiting put) are          * signalled if it ever changes from capacity. Similarly          * for all other uses of count in other wait guards.          */       	/** 判断队列是否已经满了 */         while (count.get() == capacity) {           	/** 调用添加元素锁的Condition的await方法进入等待队列进行等待，并且释放锁 */             notFull.await();         }       /** 设置尾节点的下一个节点为新创建节点     		* 将新创建节点设置为尾节点       	*/         enqueue(node);       	/** 调用getAndIncrement方法，先获取count数量，在将count队列元素加一 */         c = count.getAndIncrement();       	/** 判断当前元素小于队列最大数量则唤醒等待线程继续放入元素 */         if (c + 1 < capacity)             notFull.signal();     } finally {       	/** 释放锁 */         putLock.unlock();     }   	/** 如果队列之前为空的话，则调用signalNotEmpty方法唤醒之前等待消费的线程进行消费 */     if (c == 0)         signalNotEmpty(); }static class Node {     E item; 		/** 指向下一个节点 */     Node next;      Node(E x) { item = x; } }private void enqueue(Node node) {     // assert putLock.isHeldByCurrentThread();     // assert last.next == null;   	/** 设置尾节点的下一个节点为新创建节点     	* 将新创建节点设置为尾节点       */     last = last.next = node; }private void signalNotEmpty() {     final ReentrantLock takeLock = this.takeLock;     takeLock.lock();     try {       	/** 唤醒阻塞的消费线程 */         notEmpty.signal();     } finally {         takeLock.unlock();     } }四、take方法public E take() throws InterruptedException {     E x;     int c = -1;   	/** 获取队列中元素数量 */     final AtomicInteger count = this.count;   	/** 获取元素出队锁 */     final ReentrantLock takeLock = this.takeLock;   	/** 加一个可以被中断的锁，保证同一时间只有一个线程可以添加数据     	* 如果加锁线程被中断了，此时加锁会失败，抛出一个线程中断的异常出来       */     takeLock.lockInterruptibly();     try {       	         while (count.get() == 0) {             notEmpty.await();         }       	/** 获取第一个元素 */         x = dequeue();       	/** 调用getAndIncrement方法，先获取count数量，在将count队列元素加一 */         c = count.getAndDecrement();         if (c > 1)           	/** 如果当前队列数量大于1则进行唤醒其他等待消费线程 */             notEmpty.signal();     } finally {       	/** 释放锁 */         takeLock.unlock();     }   	/** 判断队列之前是否已经满了，满了说明可能有有线程在等待     	* 则调用signalNotFull唤醒等待线程       */     if (c == capacity)         signalNotFull();     return x; }private E dequeue() {     // assert takeLock.isHeldByCurrentThread();     // assert head.item == null;     Node h = head;   	/** 获取头节点下一个元素，因为头节点一般为空，忽略 */     Node first = h.next;   	/** 将当前头节点的下一个元素指向自己，后续会gc回收掉 */     h.next = h; // help GC   	/** 在将下一个元素设置为头节点 */     head = first;   	/** 获取数据 */     E x = first.item;   	/** 将新的头节点元素设置为空，保证头节点元素一直为空 */     first.item = null;   	/** 返回数据 */     return x; }private void signalNotFull() {     final ReentrantLock putLock = this.putLock;     putLock.lock();     try {         notFull.signal();     } finally {         putLock.unlock();     } }五、iterator方法public Iterator iterator() {     return new Itr(); }Itr() {   	/** 将队列中获取元素和添加元素的锁进行加锁 */     fullyLock();     try {         current = head.next;         if (current != null)             currentElement = current.item;     } finally {       	/** 对获取元素锁和添加元素锁进行释放 */         fullyUnlock();     } }public E next() {   	/** 将队列中获取元素和添加元素的锁进行加锁 */     fullyLock();     try {         if (current == null)             throw new NoSuchElementException();         E x = currentElement;         lastRet = current;       	/** 获取下一个元素 */         current = nextNode(current);         currentElement = (current == null) ? null : current.item;       	/** 返回元素 */         return x;     } finally {       	/** 对获取元素锁和添加元素锁进行释放 */         fullyUnlock();     } }void fullyLock() {     putLock.lock();     takeLock.lock(); }void fullyUnlock() {     takeLock.unlock();     putLock.unlock(); }private Node nextNode(Node p) {     for (;;) {         Node s = p.next;         if (s == p)             return head.next;         if (s == null || s.item != null)             return s;         p = s;     } }