You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
/** The queued items */finalObject[] items;
/** items index for next take, poll, peek or remove */inttakeIndex;
/** items index for next put, offer, or add */intputIndex;
/** Number of elements in the queue */intcount;
/** Concurrency control uses the classic two-condition algorithm* found in any textbook.*//** Main lock guarding all access */finalReentrantLocklock;
/** Condition for waiting takes */privatefinalConditionnotEmpty;
/** Condition for waiting puts */privatefinalConditionnotFull;
/*** Shared state for currently active iterators, or null if there* are known not to be any. Allows queue operations to update* iterator state.*/transientItrsitrs = null;
/*** Linked list node class*/staticclassNode<E> {
Eitem;
/** * One of: * - the real successor Node * - this Node, meaning the successor is head.next * - null, meaning there is no successor (this is the last node) */Node<E> next;
Node(Ex) { item = x; }
}
拥有成员变量如下,其中包括一个 head 节点(item 为 null,next 节点才是first 节点)和一个 last 节点(next 节点为 null),然后还包括两个锁:takeLock 和 putLock
/** The capacity bound, or Integer.MAX_VALUE if none */privatefinalintcapacity;
/** Current number of elements */privatefinalAtomicIntegercount = newAtomicInteger();
/*** Head of linked list.* Invariant: head.item == null*/transientNode<E> head;
/*** Tail of linked list.* Invariant: last.next == null*/privatetransientNode<E> last;
/** Lock held by take, poll, etc */privatefinalReentrantLocktakeLock = newReentrantLock();
/** Wait queue for waiting takes */privatefinalConditionnotEmpty = takeLock.newCondition();
/** Lock held by put, offer, etc */privatefinalReentrantLockputLock = newReentrantLock();
/** Wait queue for waiting puts */privatefinalConditionnotFull = putLock.newCondition();
方法解析
offer(E e)
源码如下,另外还有一个重载方法 offer(E e, long timeout, TimeUnit unit),在队列已满的时候,进行超时阻塞。
publicbooleanoffer(Ee) {
if (e == null) thrownewNullPointerException();
finalAtomicIntegercount = this.count;
// 队列已满则返回 falseif (count.get() == capacity)
returnfalse;
intc = -1;
// 根据待添加元素 e 创建一个 node 节点Node<E> node = newNode<E>(e);
finalReentrantLockputLock = this.putLock;
putLock.lock();
try {
if (count.get() < capacity) {
// 将 last 节点的 next 设置为 node 节点 ,然后将 node 节点设置为 last 节点enqueue(node);
c = count.getAndIncrement();
// 如果添加该元素后队列还未满,则调用 notFull.signal() 方法,即唤醒添加元素时由于队列已满导致阻塞的线程if (c + 1 < capacity)
notFull.signal();
}
} finally {
putLock.unlock();
}
if (c == 0)
// 一个获取 takeLock、notEmpty.signal()、释放 takeLock 操作过程。也就是通知其他线程可以移除元素了。signalNotEmpty();
returnc >= 0;
}
privateEdequeue() {
// assert takeLock.isHeldByCurrentThread();// assert head.item == null;Node<E> h = head;
Node<E> first = h.next;
h.next = h; // help GChead = first;
Ex = first.item;
first.item = null;
returnx;
}
put(E e)
publicvoidput(Ee) throwsInterruptedException {
if (e == null) thrownewNullPointerException();
// Note: convention in all put/take/etc is to preset local var// holding count negative to indicate failure unless set.intc = -1;
Node<E> node = newNode<E>(e);
finalReentrantLockputLock = this.putLock;
finalAtomicIntegercount = 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();
}
enqueue(node);
c = count.getAndIncrement();
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();
}
take()
publicEtake() throwsInterruptedException {
Ex;
intc = -1;
finalAtomicIntegercount = this.count;
finalReentrantLocktakeLock = this.takeLock;
takeLock.lockInterruptibly();
try {
while (count.get() == 0) {
notEmpty.await();
}
x = dequeue();
c = count.getAndDecrement();
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();
returnx;
}
remove(Object o)
因为需要操作整个链表,因此需要同时拥有两个锁才能操作。
publicbooleanremove(Objecto) {
if (o == null) returnfalse;
// putLock.lock(); takeLock.lock();fullyLock();
try {
// 遍历查找for (Node<E> trail = head, p = trail.next;
p != null;
trail = p, p = p.next) {
if (o.equals(p.item)) {
unlink(p, trail);
returntrue;
}
}
returnfalse;
} finally {
fullyUnlock();
}
}
contains(Object o)
因为需要操作整个链表,因此需要同时拥有两个锁才能操作。
publicbooleancontains(Objecto) {
if (o == null) returnfalse;
// putLock.lock(); takeLock.lock();fullyLock();
try {
// 遍历查找for (Node<E> p = head.next; p != null; p = p.next)
if (o.equals(p.item))
returntrue;
returnfalse;
} finally {
fullyUnlock();
}
}
首先先介绍一下 Queue、AbstractQueue等接口和类。
Queue
该接口针对添加元素、移除元素、获取第一个元素各有两个方法:
BlockingQueue
该接口实现了 Queue 接口,针对添加元素、移除元素各有两个方法
添加元素:
一直等待:
超时等待:
移除元素:
一直等待:
超时等待:
AbstractQueue
add(E e)
remove()
element()
ArrayBlockingQueue
简介
继承体系为:
拥有的变量如下:
线程安全实现
线程安全主要是通过可重入锁 ReentrantLock 来实现的。
add(E e)
offer(E e)
enqueue(E e)
poll()
dequeue()
阻塞
阻塞是基于 Condition 接口实现的。Condition 拥有类似的操作:await/signal。Condition 和一个 Lock 相关,由lock.newCondition() 来创建。只有当前线程获取了这把锁,才能调用 Condition 的 await 方法来等待通知,否则会抛出异常。
put(E e)
take()
同样对于take方法会有一个 notEmpty 的 Condition。
总结
LinkedBlockingQueue
简介
LinkedBlockingQueue 是使用一个链表来实现,拥有一个内部类 Node 其中 Node 结构如下:
拥有成员变量如下,其中包括一个 head 节点(item 为 null,next 节点才是first 节点)和一个 last 节点(next 节点为 null),然后还包括两个锁:takeLock 和 putLock
方法解析
offer(E e)
源码如下,另外还有一个重载方法 offer(E e, long timeout, TimeUnit unit),在队列已满的时候,进行超时阻塞。
enqueue(Node e)
poll()
和 offer(E e) 方法类似,除此之外也提供了一个重载方法 poll(long timeout, TimeUnit unit),在队列为空的时候,进行超时阻塞。
dequeue()
put(E e)
take()
remove(Object o)
因为需要操作整个链表,因此需要同时拥有两个锁才能操作。
contains(Object o)
因为需要操作整个链表,因此需要同时拥有两个锁才能操作。
总结
The text was updated successfully, but these errors were encountered: