LinkedBlockingQueue源码解析

摘要: 闲来无事 ^_^ 如有错误请指出，避免误人子弟

LinkedBlockingQueue源码解析

什么是LinkedBlockingQueue

LinkedBlockingQueue底层是由节点链表实现的定长阻塞队列(阻塞表示如果没有原始那么获取元素会阻塞当前线程)

LinkedBlockingQueue用来干嘛

LinkedBlockingQueue一般用于生产者消费者模型业务(排队机制，先进先出)

源码解析

数据的存储

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
private static final long serialVersionUID = -6903933977591709194L;
/**
* Linked list node class
*/
static class Node<E> {//存储数据的节点
E item;

Node<E> next;

Node(E x) { item = x; }
}

/** The capacity bound, or Integer.MAX_VALUE if none */
private final int capacity;//链表的最大长度，如果不设置值默认为Integer.MAX_VALUE

/** Current number of elements */
private final AtomicInteger count = new AtomicInteger(0);//统计数量线程安全

/**
* Head of linked list.
* Invariant: head.item == null
*/
private transient Node<E> head;//头节点

/**
* Tail of linked list.
* Invariant: last.next == null
*/
private transient Node<E> last;//尾节点

/** Lock held by take, poll, etc */
private final ReentrantLock takeLock = new ReentrantLock();//tackLock

/** Wait queue for waiting takes */
private final Condition notEmpty = takeLock.newCondition();//tackLock条件不为空

/** Lock held by put, offer, etc */
private final ReentrantLock putLock = new ReentrantLock();//putLock

/** Wait queue for waiting puts */
private final Condition notFull = putLock.newCondition();//putLock条件没满
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
public LinkedBlockingQueue(int capacity) {
if (capacity <= 0) throw new IllegalArgumentException();
this.capacity = capacity;
last = head = new Node<E>(null);//默认last=head=空节点
}

数据的操作

add

public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();//不能存储空元素
int c = -1;
Node<E> node = new Node(e);//创建节点
final ReentrantLock putLock = this.putLock;//获得putLock
final AtomicInteger count = this.count;//获取当前数量
putLock.lockInterruptibly();//获取锁，如果有调用Thread.Interrupted()直接抛出异常
try {

while (count.get() == capacity) {//如果当前队列以满，进入等待状态
notFull.await();
}
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
}

public boolean offer(E e, long timeout, TimeUnit unit)	offer(e)类似
throws InterruptedException {

if (e == null) throw new NullPointerException();//不能存储空元素
long nanos = unit.toNanos(timeout);//装换为纳秒
int c = -1;
final ReentrantLock putLock = this.putLock;
final AtomicInteger count = this.count;
putLock.lockInterruptibly();
try {
while (count.get() == capacity) {
if (nanos <= 0)
return false;
nanos = notFull.awaitNanos(nanos);//等待一段时间
}
enqueue(new Node<E>(e));
c = count.getAndIncrement();//递增
if (c + 1 < capacity)//如果未满唤醒notFull.awit
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();//唤醒notEmpty.await()
return true;
}
private void enqueue(Node<E> node) {
// assert putLock.isHeldByCurrentThread();
// assert last.next == null;
//拆分为两步 last.next = node，last = node
//每次head.next=当前的last然后last.next指向node
last = last.next = node;

}

remove

public boolean remove(Object o) {
if (o == null) return false;
fullyLock();//删除数据时全部lock
try {
for (Node<E> trail = head, p = trail.next;
p != null;
trail = p, p = p.next) {
if (o.equals(p.item)) {
unlink(p, trail);
return true;
}
}
return false;
} finally {
fullyUnlock();
}
}
void unlink(Node<E> p, Node<E> trail) {
// assert isFullyLocked();
// p.next is not changed, to allow iterators that are
// traversing p to maintain their weak-consistency guarantee.
p.item = null;
trail.next = p.next;//前后元素执行，大年元素设置为空
if (last == p)
last = trail;
if (count.getAndDecrement() == capacity)//count获取数量同时递减(获取数量为递减钱数量)
notFull.signal();//唤醒 notFull.await()
}

get

//获取元素，消费，可能被中断
public E take() throws InterruptedException {
E x;
int c = -1;
final AtomicInteger count = this.count;
final ReentrantLock takeLock = this.takeLock;
takeLock.lockInterruptibly();//如果有调用Thread.Interrupted()抛出异常
try {
while (count.get() == 0) {
notEmpty.await();//元素为空进入等待状态
}
x = dequeue();//
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}
//获取元素，消费
public E poll() {
final AtomicInteger count = this.count;
if (count.get() == 0)
return null;
E x = null;
int c = -1;
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
if (count.get() > 0) {
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
}
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
return x;
}
//查看元素
public E peek() {
if (count.get() == 0)
return null;
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
Node<E> first = head.next;
if (first == null)
return null;
else
return first.item;
} finally {
takeLock.unlock();
}
}
[null,aaa,bbb] queue
[null,bbb] delete after queue
去掉头部null元素获取aaa元素修改aaa元素的item=null
private E dequeue() {
// assert takeLock.isHeldByCurrentThread();
// assert head.item == null;
Node<E> h = head;
Node<E> first = h.next;//first第一个有值的节点
h.next = h; // help GC
head = first;
E x = first.item;//获取元素
first.item = null;//设置为空
return x;
}

什么时候扩容

定长链表不支持扩容

是否线程安全

线程安全

使用注意事项

默认创建方式链表醉大长度为Ineger.MAX_SIZE

引用

jdk源码