Lock和Condition在JDK中LinkedBlockingQueue的应用

import java.util.concurrent.LinkedBlockingQueue.Node;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;/**
* LinkedBlockingQueue核心方法源码分析
*
*
*/
public class LinkedBlockingQueue { /** Current number of elements */
//使用AtomicInteger的原因是：LinkedBlockingQueue的take和put使用的是两把锁。所以需要对count进行同步。
//同时count使用AtomicInteger可以解决take和put的冲突操作
private final AtomicInteger count = new AtomicInteger(0); /** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock(); /** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition(); //绑定takeLock /** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock(); /** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();//绑定putLock /**
* Inserts the specified element at the tail of this queue, waiting if
* necessary for space to become available.
*
* @throws InterruptedException
* {@inheritDoc}
* @throws NullPointerException
* {@inheritDoc}
*/
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<E> 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) {
notFull.await(); //绑定putLock
}
enqueue(node);
// 此处 c =size - 1（size为容器实际大小）
c = count.getAndIncrement();
if (c + 1 < capacity) //c+1 =size-1+1 =size ，如果c + 1 =size < capacity 的话
notFull.signal(); //唤醒其他生产者生产数据
} finally {
putLock.unlock();
}
if (c == 0) //c=size-1==0，就是size==1？如果size=1代表还有元素，通知消费者生产数据
signalNotEmpty(); //notEmpty.signal();
} /**
* Signals a waiting take. Called only from put/offer (which do not
* otherwise ordinarily lock takeLock.)
*/
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
notEmpty.signal(); //绑定takeLock
} finally {
takeLock.unlock();
}
} 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(); //绑定takeLock
}
x = dequeue();
// 有点绕：获取当前count的值，然后减1。
c = count.getAndDecrement();
//而此时由于x = dequeue()代码消费了一个数据，所以c=size+1
if (c > 1)//当前c=size+1>1的话，则就是size>0.所以此时就是有元素的，唤醒其他消费者消费数据
notEmpty.signal(); //绑定takeLock
} finally {
takeLock.unlock();
}
if (c == capacity) //c = size +1 ==capacity 所以size =capacity-1，则容器还没有满，所以通知生产者生产数据
signalNotFull(); // 调用notFull.signal();
return x;
} /**
* Signals a waiting put. Called only from take/poll.
*/
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
notFull.signal(); //绑定putLock
} finally {
putLock.unlock();
}
}
}