Queue
J.U.C中分为阻塞队里和非阻塞队列。
阻塞队列在满时进行入列操作会被阻塞，空时进行出列操作会被阻塞，很适合并发编程中最常见的生产者-消费者模式。
非阻塞队使用CAS无锁算法避免锁竞争，相比同步方式实现的队列，提高了吞吐量。
阻塞队列：
- ArrayBlockingQueue基于数组实现的有界阻塞队列。
- LinkedBlockingQueue基于链表实现的有界阻塞队列。
- PriorityBlockingQueue基于数组实现的，支持优先级排序的无界阻塞队列。
- LinkedBlockingDeque基于链表实现的双端阻塞队列。
- SynchronousQueue不存储元素的阻塞队列。
- LinkedTransferQueue基于链表实现的无界阻塞队列。
非阻塞队列：
- ConcurrentLinkedQueue基于链表实现的无界非阻塞队列。
- ConcurrentLinkedDeque基于链表实现的无界非阻塞双端队列。
队列的入列、出列方法及处理方式(阻塞和超时只适用于阻塞队列)：
| 方法\处理方式 | 异常 | 特殊值 | 阻塞 | 超时 |
|---|---|---|---|---|
| 入列方法 | add(e) | offer(e) | put(e) | offer(e, time, unit) |
| 出列方法 | remove() | poll() | take | poll(time, unit) |
| 查看方法 | element() | peek() | 无 | 无 |
ArrayBlockingQueue
基于数组实现的有界阻塞队列：
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
/** 数组容器 */
final Object[] items;
/** 出列下标 */
int takeIndex;
/** 入列下标 */
int putIndex;
/** 元素个数 */
int count;
/** 重入锁 */
final ReentrantLock lock;
/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;
public ArrayBlockingQueue(int capacity) {}
public ArrayBlockingQueue(int capacity, boolean fair) {}
public ArrayBlockingQueue(int capacity, boolean fair,
Collection<? extends E> c) {}
... ...
}
ArrayBlockingQueue没有默认长度，初始化的时候必须指定。
fair参是用来设置重入锁lock的公平性，重入锁默认是非公平锁所以不能保证线程公平的访问队列。可以通过fair将重入锁设置为公平锁，但是会降低部分吞吐量。
生产者线程和消费者线程线程的协调工作是由两个Condition完成的。
阻塞入列：
public void put(E e) throws InterruptedException {
checkNotNull(e);//元素为空抛异常
final ReentrantLock lock = this.lock;
//加锁，锁响应中断
lock.lockInterruptibly();
try {
//队列已满，入列线程在notFull上等待
while (count == items.length){
notFull.await();
}
//插入元素
insert(e);
} finally {
lock.unlock();//释放锁
}
}
private void insert(E x) {
//加入到数组
items[putIndex] = x;
//inc() putIndex下标等于capacity如果等于队列长度返回0
putIndex = inc(putIndex);
++count;//元素数量递增
//唤醒在notEmpty上等待的出列线程
notEmpty.signal();
}
阻塞出列：
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();//加锁，响应中断
try {
//队列已空，出列线程在notEmpty上等待
while (count == 0){
notEmpty.await();
}
//出列
return extract();
} finally {
lock.unlock();//解锁
}
}
private E extract() {
//数组
final Object[] items = this.items;
//元素类型泛型转换
E x = this.<E>cast(items[takeIndex]);
//置空下标为takeIndex的元素
items[takeIndex] = null;
//inc() putIndex下标等于capacity如果等于队列长度返回0
takeIndex = inc(takeIndex);
--count;//元素个数递减
//唤醒在notFull上等待的入列线程
notFull.signal();
return x;
}
当队列满时，入列的线程会阻塞在notFull上，当有出列操作时唤醒notFull上等待的线程，队列空时出列线程会阻塞在notEmpty上，当有入列操作时唤醒在notEmpty上等待的线程，典型的生产者消费者逻辑，关键点在于线程的协调。
LinkedBlockingQueue
基于单向链表实现的有界阻塞队列：
public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable{
/** 内部类 节点 */
static class Node<E> {}
/** 容量 */
private final int capacity;
/** 元素数量 计数器 */
private final AtomicInteger count = new AtomicInteger(0);
/** 头节点 */
private transient Node<E> head;
/** 尾节点 */
private transient Node<E> last;
/** 出列锁 */
private final ReentrantLock takeLock = new ReentrantLock();
/** takeLock->condition */
private final Condition notEmpty = takeLock.newCondition();
/** 入列锁 */
private final ReentrantLock putLock = new ReentrantLock();
/** putLock->condition */
private final Condition notFull = putLock.newCondition();
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
public LinkedBlockingQueue(int capacity) {}
public LinkedBlockingQueue(Collection<? extends E> c) {}
... ...
}
LinkedBlockingQueue在初始化时可以不指定长度，默认长为整数的最大值 2147483647 。
使用了两把锁对对出列和入列进行了锁分离，takeLock出列锁、putLock入列锁。
LinkedBlockingQueue没有公平性设置，只能使用非公平锁。
阻塞入列：
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;
//元素数量
final AtomicInteger count = this.count;
putLock.lockInterruptibly();//加锁，响应中断
try {
//队列已满,入列线程在notFull上等待
while (count.get() == capacity) {
notFull.await();
}
//入列，在尾节点后链入node
enqueue(node);
//获取元素数量 后加1
c = count.getAndIncrement();
//如果队列还没满
//唤醒在notFull等待的入列线程，表示可继续入列
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();//解锁
}
//原本为空的队列，即使加入一个元素
//唤醒在notEmpty上等待的出列线程
if (c == 0)
signalNotEmpty();
}
//链入尾节点
private void enqueue(Node<E> node) {
last = last.next = node;
}
//唤醒在notEmpty上等待的出列线程
private void signalNotEmpty() {
//出列锁
final ReentrantLock takeLock = this.takeLock;
takeLock.lock();
try {
//唤醒
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 {
//队列为空，出列线程在notEmpty上等待
while (count.get() == 0) {
notEmpty.await();
}
//出列
x = dequeue();
//获取元素数量 后减1
c = count.getAndDecrement();
//出列之后，队列还没空，表示可继续出列
//唤醒在notEmpty等待的出列线程
if (c > 1)
notEmpty.signal();
} finally {
takeLock.unlock();
}
//队列有一个空位，唤醒入列线程
if (c == capacity)
signalNotFull();
return x;
}
//唤醒入列线程
private void signalNotFull() {
//入列锁
final ReentrantLock putLock = this.putLock;
putLock.lock();
try {
notFull.signal();
} finally {
putLock.unlock();
}
}
当队列满时，入列的线程会阻塞在notFull上，当有出列操作时唤醒notFull上等待的线程，队列空时出列线程会阻塞在notEmpty上，当有入列操作时唤醒在notEmpty上等待的线程，入列和出列使用了两把锁，唤醒notFull时要在putLock监视范围，唤醒notEmpty要做takeLock的监视范围。
ArrayBlockingQueue和LinkedBlockingQueue的差异
- ArrayBlockingQueue使用循环数组必须指定容量，LinkedBlockingQueue使用链表可以不指定容量，能预判队列容量使用ArrayBlockingQueue可以更有效的利用内存。LinkedBlockingQueue如果没有指定容量，过快大批量的入列有可能会导致内存溢出。
- ArrayBlockingQueue可以设置为公平锁，使得线程能够公平地访问队列。
- LinkedBlockingQueue使用锁分离，入列和出列使用不同的锁，之间互不干扰，减少了锁争用的次数，吞吐量比ArrayBlockingQueue更高。
PriorityBlockingQueue
基于数组实现的无界阻塞队列，因为是无界队列当数组长度不够时会自动扩容所以put方法不会阻塞，但是队列空时进行take会阻塞。
PriorityBlockingQueue不再是FIFO，而是根据元素的排序来确定元素出列的优先级，元素必须实现Comparable接口。
LinkedBlockingDeque
基于链表实现的组成的双向阻塞队列，同时支持FIFO和FILO两种操作方式。
SynchronousQueue
SynchronousQueue是一个没有容器的队列，所谓没有容器就是指它不能存储任何元素。不像ArrayBlockingQueue或LinkedBlockingQueue如果队列没有满，生产线程入列之后就返回了，而SynchronousQueue不同，因为它没有缓冲存储区所以生产者线程入列之后会一直阻塞，直到有消费线程取走数据。
就像一手交钱一手交货的过程，卖方拿着货物不松手，直到买房把钱给他，买方也是一样的拿着钱不松手，直到卖方把货物给他。
所以SynchronousQueue从线程的角度看是一个配对的过程一个生成线程必须匹配一个消费线程，一个消费线程必须匹配一个生成线程，从数据的角度看是一个数据传递的过程生成线程将数据传递给消费线程。
SynchronousQueue：
public class SynchronousQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable{
/** Transferer */
abstract static class Transferer {
abstract Object transfer(Object e, boolean timed, long nanos);
}
/** Transferer子类 栈 */
static final class TransferStack extends Transferer {}
/** Transferer子类 队列 */
static final class TransferQueue extends Transferer {}
/** transferer实例 */
private transient volatile Transferer transferer;
/** 默认构造 */
public SynchronousQueue() {
this(false);
}
/** fair 公平性参数 */
public SynchronousQueue(boolean fair) {
transferer = fair ? new TransferQueue() : new TransferStack();
}
... ...
}
SynchronousQueue可以设置公平性策略，默认是非公平队列。
Transferer是核心设置，实现线程数据传递的基础，公平性队列用TransferQueue新入列的节点会在队尾或者和队头节点批量，非公平队列用TransferStack新入列的节点会在栈顶进行匹配。因为没有缓冲存储所以容器类的常用方法size()、contains(Object o)、remove(Object o)等对其来说根本没用。
TransferStack：
static final class TransferStack extends Transferer {
/** 消费端 consumer */
static final int REQUEST = 0;
/** 生成端 producer */
static final int DATA = 1;
/** 匹配 */
static final int FULFILLING = 2;
/** true 匹配中 */
static boolean isFulfilling(int m) {
return (m & FULFILLING) != 0;
}
/** TransferStacks节点 */
static final class SNode {}
/** 栈顶节点 */
volatile SNode head;
/** CAS设置栈顶 */
boolean casHead(SNode h, SNode nh) {}
/** 构造节点 */
static SNode snode(SNode s, Object e, SNode next, int mode) {}
/** 交换方法 */
Object transfer(Object e, boolean timed, long nanos) {}
/** 线程等待 */
SNode awaitFulfill(SNode s, boolean timed, long nanos){}
/** 是否自旋 */
boolean shouldSpin(SNode s){}
/** 将节点从栈清除 */
void clean(SNode s){}
//Unsafe 相关初始化 ... ..
}
SNode ：
static final class SNode {
volatile SNode next; //后继节点
volatile SNode match; //匹配节点
volatile Thread waiter; //等待线程
Object item; //生产端：data;消费端：null
int mode;//模式:DATA/REQUEST/FULFILLING
SNode(Object item) {
this.item = item;
}
/** CAS设置后继节点 */
boolean casNext(SNode cmp, SNode val) {
return cmp == next
&& UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
}
/** 当前节点和s节点匹配，匹配成功唤醒当前节点等待的线程 */
boolean tryMatch(SNode s) {
//当前节点的match设置为s
if (match == null
&&UNSAFE.compareAndSwapObject(this, matchOffset, null, s)) {
Thread w = waiter;
if (w != null) { //waiter不为空 唤醒。
waiter = null;
LockSupport.unpark(w);
}
return true;
}
//如果match == s则说明已经匹配成功
return match == s;
}
//取消 将match设置为自身
void tryCancel() {
UNSAFE.compareAndSwapObject(this, matchOffset, null, this);
}
//是否已取消
boolean isCancelled() {
return match == this;
}
// Unsafe mechanics ... ...
}
TransferStack的transfer方法：
/** Puts 和 takes 数据交换 */
Object transfer(Object e, boolean timed, long nanos) {
SNode s = null;
// 0消费端，1生产端
int mode = (e == null) ? REQUEST : DATA;
for (;;) {
SNode h = head;// 头节点
// 栈为空,当前线程进入等待
// 或者栈不为空，但是栈顶元素模式与当前线程模式相同
// 即同为生成着或消费者，比如线程put线程
// 当前线程进入等待
if (h == null || h.mode == mode) {
if (timed && nanos <= 0) { // 不等待
//h不为空并被取消
if (h != null && h.isCancelled())
//出栈
casHead(h, h.next);
else
return null;
// 压栈 更新栈顶为s
} else if (casHead(h, s = snode(s, e, h, mode))) {
// 进入等待，等待一个互补的节点进行匹配
SNode m = awaitFulfill(s, timed, nanos);
// 取消的时候将match设置成了this
// 所以m==s即被取消，清除，返回。
if (m == s) {
clean(s);
return null;
}
//已经完成了批量
if ((h = head) != null && h.next == s) {
casHead(h, s.next);
}
// 如果是消费者则返回生成值的值
// 如果是生产者返回自身的值
return (mode == REQUEST) ? m.item : s.item;
}
// 栈顶和当前节点互补即模式不同，进入匹配逻辑
} else if (!isFulfilling(h.mode)) {
if (h.isCancelled()) {// 已取消，出栈，置换栈顶为h.next
casHead(h, h.next);
//构造当前“正在匹配"状态的节点s
} else if (casHead(h, s = snode(s, e, h, FULFILLING | mode))) {
for (;;) { // 循环直到找到一个可以匹配的节点
SNode m = s.next; // m即与s匹配的节点
//m==null说明栈s之后无元素，可能被其他线程匹配了。
//s出栈，s置空，进行最外层的循环.
if (m == null) {
casHead(s, null);
s = null;
break;
}
//mn为后备的栈顶
//匹配成功，将s和m同时出栈，mn为栈顶
SNode mn = m.next;
if (m.tryMatch(s)) {
//匹配成功，mn设置为栈顶
casHead(s, mn);
// 如果是消费者则返回生成值的值
// 如果是生产者返回自身的值
return (mode == REQUEST) ? m.item : s.item;
} else
// 设置匹配失败，则说明m已经被其他节点匹配了
s.casNext(m, mn); // help unlink
}
}
} else { // 非栈顶匹配，逻辑与栈顶匹配一致
SNode m = h.next; // m is h's match
if (m == null) // waiter is gone
casHead(h, null); // pop fulfilling node
else {
SNode mn = m.next;
if (m.tryMatch(h)) // help match
casHead(h, mn); // pop both h and m
else // lost match
h.casNext(m, mn); // help unlink
}
}
}
}
// 等待
SNode awaitFulfill(SNode s, boolean timed, long nanos) {
long lastTime = timed ? System.nanoTime() : 0;
// 当前线程
Thread w = Thread.currentThread();
// 头节点
SNode h = head;
// 自旋次数
int spins = (shouldSpin(s) ?
(timed ? maxTimedSpins : maxUntimedSpins) : 0);
for (;;) {
if (w.isInterrupted()) {// 当前线程中断
s.tryCancel();//取消节点
}
SNode m = s.match;
if (m != null) {//匹配成功，返回匹配的节点
return m;
}
// 超时
if (timed) {
long now = System.nanoTime();
nanos -= now - lastTime;
lastTime = now;
if (nanos <= 0) {
s.tryCancel();//取消
continue;
}
}
// 自旋，直到spins==0，进入等待，自旋的目的是为了减少线程挂起的次数
// 如果线程挂起前，匹配线程来了，则线程不需要挂起
if (spins > 0) {
spins = shouldSpin(s) ? (spins - 1) : 0;
}
// 设置节点的等待线程
else if (s.waiter == null) {
s.waiter = w; // establish waiter so can park next iter
}
// 挂起操作
else if (!timed) {
LockSupport.park(this);
} else if (nanos > spinForTimeoutThreshold) {
LockSupport.parkNanos(this, nanos);
}
}
}
TransferStack的transfer大致逻辑是：线程A进行put(A)，此时栈为空，将节点A入栈，线程A挂起，栈顶为节点A。线程B进行put(B),和节点A模式一样，同为DATA,将节点B入栈线程B挂起，栈顶为节点B。线程C进行take(),和栈顶B模式互补，将节点C的状态设置为FULFILLING入栈，开始进行匹配操作，匹配成则线程B被唤醒、节点B和节点C出栈，并返回节点B的值。
如果节点B和节点C正在匹配中，即栈顶节点的状态为ULFILLING，线程D进行take(),那么线程D将帮助节点B和节点C完成匹配和出栈，自己在留在下一轮循环中匹配。
线程A是先入栈的反而后匹配，所以TransferStack的匹配过程是非公平的。TransferQueue则是在队尾入列，从队列头匹配，能保证先入列的线程可以尽早的得到匹配，阻塞和匹配逻辑和上述差不多，只是入列过程不一样，不再赘述。
SynchronousQueue不能使用在缓冲场景，但是非常适合用在传递场景，由于其阻塞过程没有锁的竞争吞吐量高于ArrayBlockingQueue和LinkedBlockingQueue。
LinkedTransferQueue
LinkedTransferQueue基于链表的无界阻塞队列。同时它还实现了TransferQueue接口，这是一个在JDK1.7中新增的接口，接口中的transfer系列方法会使生产者一直阻塞直到所添加到队列的元素被某一个消费者所消费。和SynchronousQueue中实现的TransferQue意思差不多。
与SynchronousQueue相比LinkedTransferQueue的应用更广泛，可以使用put/take方法用作缓冲，还可以使用transfer方法用作传递，可以看做是ConcurrentLinkedQueue、SynchronousQueue（公平模式）和LinkedBlockingQueue的超集。
ConcurrentLinkedQueue
使用单向链表构造的无界非阻塞队列：
public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
implements Queue<E>, java.io.Serializable {
//内部类 节点
private static class Node<E> {
volatile E item;
volatile Node<E> next;
... ...
}
//头节点
private transient volatile Node<E> head;
//尾节点
private transient volatile Node<E> tail;
public ConcurrentLinkedQueue() {
head = tail = new Node<E>(null);
}
public ConcurrentLinkedQueue(Collection<? extends E> c){}
... ...
}
入列操作：
public boolean offer(E e) {
checkNotNull(e);// 检查e是否为空，为空直接抛异常
// 构造新节点
final Node<E> newNode = new Node<E>(e);
// 循环，移动p节点、确保CAS操作成功
for (Node<E> t = tail, p = t;;) {
Node<E> q = p.next;// p的后继节点
if (q == null) {// q为空，说明p为尾节点
// CAS更新p的next节点为新入节点
if (p.casNext(null, newNode)) {
// 当p为尾节点时只进行了p.casNext()操作，
// 并没有移动尾节点。p和t中间至少隔了一个节点。
if (p != t) {
// CAS 更新尾节点
casTail(t, newNode);
}
return true;
}
} else if (p == q) {// 尾节点被出列
p = (t != (t = tail)) ? t : head;
} else {// p节点后移
p = (p != t && t != (t = tail)) ? t : q;
}
}
}
出列操作：
public E poll() {
restartFromHead: for (;;) {//循环体，移动p节点、确保CAS成功
for (Node<E> h = head, p = h, q;;) {
E item = p.item;
// 头节点的内容不为空，将其置空
if (item != null && p.casItem(item, null)) {
// 出列时，进行了p.casItem()但并没有移动头节点
// p节点和h节点中间至少隔了一个节点
if (p != h) {
// 设置头节点
updateHead(h, ((q = p.next)
!= null) ? q : p);
}
return item;
} else if ((q = p.next) == null) {//空队为空
updateHead(h, p);
return null;
} else if (p == q) {//从队列头重新开始
continue restartFromHead;
} else {// p后移
p = q;
}
}
}
}
入列操作offer(E e)只做了两件事情，第一是将新节点链到队列尾部，第二是定位尾节点将其指向新入列的节点，这两个操作都是使用CAS方式，出列poll()操作也类似。入列和出列都只需要动用一个节点，并且是无锁的，所以ConcurrentLinkedQueue在并发环境下出列和入列效率极高。
获取长度的方法，需要遍历整个链表，效率极低，所以慎用。如果想实时获取列表长度，不妨使用一个AtomicInteger在入列和出列时记录下，好过整表遍历。
public int size() {
int count = 0;
//从头节点开始遍历，p!=null说明p还没到队列尾
for (Node<E> p = first(); p != null; p = succ(p))
if (p.item != null)
// Collection.size() spec says to max out
if (++count == Integer.MAX_VALUE)
break;
return count;
}
ConcurrentLinkedDeque
使用双向链表构造的无界非阻塞双端队列，双端队列中的元素可以从两端弹出，入列和出列操作可以在表的两端进行，支持FIFO和FILO两种出列模式。
尽管看起来比队列更灵活，但实际上在应用中远不及队列有用。
码字不易，转载请保留原文连接https://www.jianshu.com/p/0120984dea81
