LinkedBlockingQueue是一个基于链表的阻塞队列，实现了BlockingQueue、java.io.Serializable接口继承自AbstractQueue
创建：
//当队列中没有任何元素的时候,此时队列的头部就等于队列的尾部,指向的是同一个节点,并且内容为null
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
last = head = new Node<E>(null);
}
//无参构造，默认使用int的最大值作为capacity的值
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
//将一个集合中的元素全部入队列。
public LinkedBlockingQueue(Collection<? extends E> c) {
this(Integer.MAX_VALUE);
final ReentrantLock putLock = this.putLock;
//获取锁
putLock.lock(); // Never contended, but necessary for visibility
try {
//迭代集合中的每一个元素,让其入队列,并且更新一下当前队列中的元素数量
int n = 0;
for (E e : c) {
if (e == null)
throw new NullPointerException();
if (n == capacity)
throw new IllegalStateException("Queue full");
//参考下面的enqueue分析
enqueue(new Node<E>(e));
++n;
}
count.set(n);
} finally {
//释放锁
putLock.unlock();
}
}
数据元素操作：
//我们看到是一个链表，里面每个节点都是一个内部类Node,所有的元素都通过Node类来进行存储
static class Node<E> {
E item;
Node<E> next;
Node(E x) { item = x; }
}
//创建一个节点，并加入链表尾部，入队
private void enqueue(Node<E> node) {
//把node赋给当前的最后一个节点的下一个节点，然后在将node设为最后一个节点
last = last.next = node;
}
//出队
private E dequeue() {
Node<E> h = head;//获取头节点：x==null
Node<E> first = h.next;//将头节点的下一个节点赋值给first
h.next = h; // 将当前将要出队的节点置null（为了使其做head节点做准备
head = first;//将当前将要出队的节点作为了头节点
E x = first.item;//获取出队节点的值
first.item = null;//将出队节点的值置空
return x;
}
重要参数：
//AtomicInteger 一个提供原子操作的Integer的类，用来计算现在队列有多少个元素
//解决在入队或出队时并发修改元素数量
private final AtomicInteger count = new AtomicInteger(0);
//元素出队入队时线程所获取的重入锁
private final ReentrantLock takeLock = new ReentrantLock();
private final ReentrantLock putLock = new ReentrantLock();
//当队列为空时，让从队列中获取元素的线程处于挂起状态
private final Condition notEmpty = takeLock.newCondition();
//当队列为空时，让元素入队列的的线程处于挂起状态
private final Condition notFull = putLock.newCondition();
入队：
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
int c = -1;
Node<E> node = new Node(e);
//当成员变量被其他线程改变时，因为在方法内部重新引用并用final修饰，保证在一次操作内数据是一致的？
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
//执行可中断的锁获取操作
putLock.lockInterruptibly();
try {
//当队列的容量到底最大时,此时线程将处于等待状态 ？为什么要使用while
while (count.get() == capacity) {
notFull.await();
}
//让元素排入队列的末尾
enqueue(node);
//更新队列中的元素个数.此处的count.getAndIncrement()方法会更新元素个数并返回旧值
c = count.getAndIncrement();
//如果添加元素后的容量，还小于指定容量，说明至少还可以再插一个元素
if (c + 1 < capacity)
notFull.signal();
} finally {
//释放锁
putLock.unlock();
}
//注意！！！这个c的值就count容量的旧值，c == 0时说明之前的队列是空队列，即出队列=的线程都处于等待状态
//上边增加了一个新的元素，队列不为空，就会唤醒正在等待获取元素的线程
if (c == 0)
signalNotEmpty();
}
//唤醒正在等待获取元素的线程
private void signalNotEmpty() {
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
//通过notEmpty唤醒获取元素的线程
notEmpty.signal();
} 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();
}
//完成元素的出队列
x = dequeue();
//更新队列中的元素个数.
c = count.getAndDecrement();
//当前队列中元素数量大于1，唤醒线程，继续执行出队操作
if (c > 1)
notEmpty.signal();
} finally {
//释放锁
takeLock.unlock();
}
//c==capaitcy的时候，在执行此次出队操作完成之前队列已经满了，去唤醒入队操作的线程进行插入操作
if (c == capacity)
signalNotFull();
return x;
}
private void signalNotFull() {
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
notFull.signal();
} finally {
putLock.unlock();
}
}
补充：
AtomicInteger
对于一个Integer线程安全的操作。
//当前值+1，采用无限循环，直到+1成功为止
//返回的是旧值
public final int getAndIncrement() {
for (;;) {
int current = get();//获取当前值
int next = current + 1;//当前值+1
if (compareAndSet(current, next))//基于CAS赋值
return current;
}
}
//当前值-1，采用无限循环，直到-1成功为止
//返回旧值
public final int getAndDecrement() {
for (;;) {
int current = get();
int next = current - 1;
if (compareAndSet(current, next))
return current;
}
}
compareAndSet这个方法多见于并发控制中，简称CAS(Compare And Swap)，意思是如果valueOffset位置包含的值与expect值相同，则更新valueOffset位置的值为update，并返回true，否则不更新，返回false。
public final boolean compareAndSet(int expect, int update) {
return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}
网友评论