SynchronousQueue
Java 6 的并发编程包中的SynchronousQueue是一个没有数据缓冲的BlockingQueue，生产者线程对其的插入操作put必须等待消费者的移除操作take，反过来也一样。不像ArrayBlockingQueue或LinkedListBlockingQueue，SynchronousQueue内部并没有数据缓存空间，你不能调用peek()方法来看队列中是否有数据元素，因为数据元素只有当你试着取走的时候才可能存在，不取走而只想偷窥一下是不行的，当然遍历这个队列的操作也是不允许的。队列头元素是第一个排队要插入数据的线程，而不是要交换的数据。数据是在配对的生产者和消费者线程之间直接传递的，并不会将数据缓冲数据到队列中。可以这样来理解：生产者和消费者互相等待对方，握手，然后一起离开。
SynchronousQueue的一个使用场景是在线程池里。Executors.newCachedThreadPool()就使用了SynchronousQueue，这个线程池根据需要（新任务到来时）创建新的线程，如果有空闲线程则会重复使用，线程空闲了60秒后会被回收。
SynchronousQueue是一种很特别的BlockingQueue，任何一个添加元素的操作都必须等到另外一个线程拿走元素才会结束。也就是SynchronousQueue本身不会存储任何元素，相当于生产者和消费者手递手直接交易。
SynchronousQueue有一个fair选项，如果fair为true，称为fair模式，否则就是unfair模式。
在fair模式下，所有等待的生产者线程或者消费者线程会按照开始等待时间依次排队，然后按照等待先后顺序进行匹配交易。这种情况用队列实现。
在unfair模式下，则刚好相反，后来先匹配，这种情况用栈实现。
public SynchronousQueue(boolean fair) {
transferer = fair ? new TransferQueue() : new TransferStack();
}因为添加元素和拿走元素是类似手递手交易的，所以对于拿走元素和添加元素操作，SynchronousQueue调用的是Transferer同一个方法transfer。
当object为null时表示是拿走元素，用于消费者线程，否则则是添加元素，用于生产者线程。因此transfer方法是分析的重点。
abstract Object transfer(Object e, boolean timed, long nanos);首先来看用于fair模式的TransferQueue的transfer方法：
看代码之前，来理一下逻辑：
开始队列肯定是空。
线程进入队列，如果队列是空的，那么就添加该线程进入队列，然后进行等待（要么有匹配线程出现，要么就是该请求超时取消）
第二个线程进入，如果前面一个线程跟它属于不同类型，也就是说两者是可以匹配的，那么就从队列删除第一个线程。
如果是相同的线程，那么做法参照2。
理清了基本逻辑，也就是会有两种情况：
队列为空或者队列中的等待线程是相同类型
队列中的等待线程是匹配的类型
Object transfer(Object e, boolean timed, long nanos) {
QNode s = null;
// e不是null表示是生成者线程，e就是产品，反之就是消费者线程
boolean isData = (e != null);
for (;;) {
QNode t = tail;
QNode h = head;
// tail和head在队列创建时会被初始化成一个虚拟节点
// 因此发现没有初始化，重新循环等待直到初始化完成
if (t == null || h == null)
continue;
// 队列为空或等待线程类型相同（不同类型才能匹配）
// 这两种情况都要把当前线程加入到等待队列中
if (h == t || t.isData == isData) {
QNode tn = t.next;
// tail对象已经被更新，出现不一致读的现象，重新循环
if (t != tail)
continue;
// 添加线程到等待队列时会先更新当前tail的next，然后
// 更新tail本身，因此出现只有next被更新的情况，应该
// 更新tail，然后重新循环
if (tn != null) {
advanceTail(t, tn);
continue;
}
// 设定了超时，剩余等待时间耗尽的时候，就无需再等待
if (timed && nanos <= 0)
return null;
// 首次使用s的时候，新建一个节点保存当前线程和数据来初始化s
if (s == null)
s = new QNode(e, isData);
// 尝试更新tail的next，把新建节点添加到tail的后面，如果失败了，就重新循环
if (!t.casNext(null, s))
continue;
// 把新建的节点设置为tail
advanceTail(t, s);
// 等待匹配线程，成功匹配则返回的匹配的值
// 否则返回当前节点，因此s和x相同表示请求被取消
Object x = awaitFulfill(s, e, timed, nanos);
if (x == s) {
clean(t, s);
return null;
}
// 这个时候已经匹配成功了，s应该是排在第一个的等待线程
// 如果s依然在队列中，那么需要更新head。
// 更新head的方法是把s这个排在第一位的节点作为新的head
// 因此需要重置一些属性使它变成虚拟节点
if (!s.isOffList()) {
advanceHead(t, s);
if (x != null)
s.item = s;
s.waiter = null;
}
// x不为null表示拿到匹配线程的数据（消费者拿到生产者的数据），
// 因此返回该数据，否则返回本身的数据（生成者返回自己的数据）
return (x != null) ? x : e;
} else { // 线程可以匹配
// 因为是队列，因此匹配的是第一个节点
QNode m = h.next;
// 同样需要检查不一致读的情况
if (t != tail || m == null || h != head)
continue;
Object x = m.item;
// 匹配失败时，把m从队列中移走，重新循环
if (isData == (x != null) || // m已经被匹配了
x == m || // m已经被取消了
!m.casItem(x, e)) { // 用CAS设置m的数据为null
advanceHead(h, m);
continue;
}
// 匹配成功，更新head
advanceHead(h, m);
// 解除m的线程等待状态
LockSupport.unpark(m.waiter);
// 返回匹配的数据
return (x != null) ? x : e;
}
}
}接着来用于Unfair模式的TransferStack的transfer方法
大体逻辑应该是一样的，不同就是队列的入队和出队操作对应到栈时就是入栈和出栈的操作。
Object transfer(Object e, boolean timed, long nanos) {
SNode s = null;
int mode = (e == null) ? REQUEST : DATA;
for (;;) {
SNode h = head;
// 栈为空或者节点类型相同的情况
if (h == null || h.mode == mode) {
if (timed && nanos <= 0) {
// 检查栈顶节点是否已经取消，如果已经取消，弹出节点
// 重新循环，接着检查新的栈顶节点
if (h != null && h.isCancelled())
casHead(h, h.next);
else
return null;
// 新建节点，并且尝试把新节点入栈
} else if (casHead(h, s = snode(s, e, h, mode))) {
// 等待匹配，如果发现是被取消的情况，则释放节点，返回null
SNode m = awaitFulfill(s, timed, nanos);
if (m == s) {
clean(s);
return null;
}
// 如果匹配的成功两个节点是栈顶的两个节点
// 把这两个节点都弹出
if ((h = head) != null && h.next == s)
casHead(h, s.next); // help s's fulfiller
return (mode == REQUEST) ? m.item : s.item;
}
} else if (!isFulfilling(h.mode)) { // 栈顶节点没有和其他线程在匹配，可以匹配
if (h.isCancelled()) // 栈顶节点的请求已经被取消
casHead(h, h.next); // 移除栈顶元素重新循环
// 尝试把该节点也入栈，该节点设置为正在匹配的状态
// 也就是isFulfilling返回true
else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {
for (;;) {
// 栈顶节点（当前线程的节点）和它的下一个节点进行匹配，m为null意味着
// 栈里没有其他节点了，因为前面该节点入栈了，需要弹出这个节点重新循环
SNode m = s.next;
if (m == null) {
casHead(s, null);
s = null;
break;
}
// 这个时候是有节点可以匹配的，尝试为这两个节点做匹配
SNode mn = m.next;
// m和s匹配成功，弹出这两个节点，返回数据；匹配失败，把m移除
if (m.tryMatch(s)) {
casHead(s, mn);
return (mode == REQUEST) ? m.item : s.item;
} else
s.casNext(m, mn);
}
}
// 栈顶正在匹配，参见代码：
// else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {
// 做法基本类似，只是这里帮助其他线程匹配，无论成功与否
// 都要重新循环
} else {
SNode m = h.next;
if (m == null)
casHead(h, null);
else {
SNode mn = m.next;
if (m.tryMatch(h))
casHead(h, mn);
else
h.casNext(m, mn);
}
}
}
}Semaphore
从代码实现的角度来说，信号量与锁很类似，可以看成是一个有限的共享锁，即只能被有限数量的线程使用的共享锁。
因为存在计数，因此Semaphore的构造函数有参数permits来设定计数：
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}涉及到线程排队等待的问题，Semaphore也支持fair和unfair模式：
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}说到线程排队，前面在说“锁”的时候讲过AbstractQueuedSynchronizer，它实现了类似获取锁失败，管理等待的线程的功能。因此信号量的实现同样需要借助这个类。
abstract static class Sync extends AbstractQueuedSynchronizer
// Unfair模式的实现
static final class NonfairSync extends Sync
// Fair模式的实现
static final class FairSync extends SyncSync类使用AbstractQueuedSynchronizer的state来存储信号量的计数：
Sync(int permits) {
setState(permits);
}因为信号量与共享锁类似，因此在获取资源和释放资源的时候使用的都是AbstractQueuedSynchronizer的shared类型的方法。
再次回到前面的unfair和fair模式，这种所谓的公平体现在获取锁的时候：unfair是后来先得，fair是先来先得。来看两者的尝试获取资源的方法：
// unfair模式
final int nonfairTryAcquireShared(int acquires) {
// 直接检查是不是有资源，根本不看前面有没有其他排队的
for (;;) {
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
// fair模式
protected int tryAcquireShared(int acquires) {
for (;;) {
// 先看看有没有排队的
if (hasQueuedPredecessors())
return -1;
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}对于信号量来说，获取资源的过程，就是一个更新资源计数的过程。对于释放资源来说，也是一样。
protected final boolean tryReleaseShared(int releases) {
for (;;) {
int current = getState();
int next = current + releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))
return true;
}
}Exchanger
Exchanger可以看做双向数据传输的SynchronousQueue，即没有生产者和消费者之分，任意两个线程都可以交换数据。
在JDK5中Exchanger被设计成一个容量为1的容器，存放一个等待线程，直到有另外线程到来就会发生数据交换，然后清空容器，等到下一个到来的线程。
从JDK6开始，Exchanger用了类似ConcurrentMap的分段思想，提供了多个slot，增加了并发执行时的吞吐量。
Exchanger不存在公平不公平的模式，因为没有排队的情况发生，只要有两个线程就可以发生数据交换。
private Object doExchange(Object item, boolean timed, long nanos) {
Node me = new Node(item);
// index是线程ID的hash值映射到0到max之间的一个值
// 一般情况下max为0，这样线程交换数据只会使用第一个slot，
// 即index是0，而max不为0情况请看下面的循环
int index = hashIndex();
// CAS操作失败的次数
int fails = 0;
for (;;) {
// 当前slot中存储的对象，也就是Node
Object y;
Slot slot = arena[index];
// 延迟加载，即只有当slot为null时才创建一个slot
// 延迟加载后重新循环一次
if (slot == null)
createSlot(index);
// slot中有数据，也就意味着有线程在等待交换数据
// 这时可以尝试用CAS重置slot(把slot存储的对象设为null)
// 用slot中存储的对象和当前线程进行数据交换
// 如果交换成功就通知原先等待的线程
else if ((y = slot.get()) != null &&
slot.compareAndSet(y, null)) {
Node you = (Node)y;
if (you.compareAndSet(null, item)) {
LockSupport.unpark(you.waiter);
return you.item;
}
// 如果slot存储的对象已经被重置为null，但是数据交换失败了
// 这时就意味着这个等待的线程的交换请求被取消了
// 在分析wait类型的方法代码时会看到如何处理这种情况
}
// 如果slot中没有存储对象，那么首先尝试把当前线程存储到slot中
// 如果存储失败了，就重新循环
else if (y == null &&
slot.compareAndSet(null, me)) {
// index为0意味着仅仅有当前线程在等待交换数据,因此直接等待即可
if (index == 0)
return timed ?
awaitNanos(me, slot, nanos) :
await(me, slot);
// 所谓的spin wait：就是固定次数循环，每次计数减一
// 对于单核系统来说，spin wait是不做的，因为单核
// 做wait时需要占用CPU，其他线程是无法使用CPU，因此这样
// 的等待毫无意义。而多核系统中spin值为2000，也就是会做
// 2000次循环。
// 如果循环完成后依然没有得到交换的数据，那么会返回一个
// CANCEL对象表示请求依旧被取消，并且把Node从slot中清除
Object v = spinWait(me, slot);
if (v != CANCEL)
return v;
// 如果取消了，就新建一个Node取消原先取消的Node用于下次循环
me = new Node(item);
int m = max.get();
// index除2，缩小slot的范围
// 同时如果m过大，减小m
if (m > (index >>>= 1))
max.compareAndSet(m, m - 1);
}
// 允许CAS失败两次，因为两个else if中都有CAS，因此这里
// 允许两个else if的CAS操作都失败过
else if (++fails > 1) {
int m = max.get();
// 失败超过3次，增大m，并且从m处重新索引
if (fails > 3 && m < FULL && max.compareAndSet(m, m + 1))
index = m + 1;
// 当index小于0，回到m，重新循环
else if (--index < 0)
index = m;
}
}
}PriorityBlockingQueue
在ScheduledThreadPoolExecutor曾经重点讲到了DelayedWorkQueue，这里说的PriorityBlockingQueue其实是DelayedWorkQueue的简化版本，实现了按序排列元素的功能。也就是说PriorityBlockingQueue是维护一个按序排列的队列，排序的方法可以通过指定Comparator来比较元素的大小，或者元素类型本身实现了Comparable接口。
因此PriorityBlockingQueue也是使用基于数组的二叉堆来实现的。
public boolean offer(E e) {
if (e == null)
throw new NullPointerException();
final ReentrantLock lock = this.lock;
// 锁定这个队列
lock.lock();
int n, cap;
Object[] array;
// 如果数组已满，则尝试为数组扩容
while ((n = size) >= (cap = (array = queue).length))
tryGrow(array, cap);
try {
Comparator<? super E> cmp = comparator;
if (cmp == null)
// 没有comparator情况下，元素类型必须实现Comparable接口
// 使用compare方法进行比较，然后插入元素到堆中
siftUpComparable(n, e, array);
else
// 制定comparator的情况下，插入元素使用comparator
// 比较元素，然后插入元素到堆中
siftUpUsingComparator(n, e, array, cmp);
size = n + 1;
notEmpty.signal();
} finally {
lock.unlock();
}
return true;
}这里看一下tryGrow，在对数组进行扩容时释放了主锁的，因为分配空间本身是不需要主锁的，只有更新数组时才会要主锁。这样可以提高并发执行的性能，减少阻塞。
private void tryGrow(Object[] array, int oldCap) {
// 扩容数组时，释放主锁，这样其他取走元素的操作就可以正常
// 操作了。这里使用一个简单的allocationSpinLock作为锁，
// 它的值为1表示锁正在被使用，为0表示锁为被占用。
// 在获取该锁时，用的CAS操作，而释放时，因为锁已经占用，
// 直接赋值为0即可。
// 分配空间本身是用不到主锁的，只有更新数组的时候才需要。
lock.unlock();
Object[] newArray = null;
if (allocationSpinLock == 0 &&
UNSAFE.compareAndSwapInt(this, allocationSpinLockOffset,
0, 1)) {
try {
int newCap = oldCap + ((oldCap < 64) ?
(oldCap + 2) :
(oldCap >> 1));
if (newCap - MAX_ARRAY_SIZE > 0) {
int minCap = oldCap + 1;
// 整数溢出
if (minCap < 0 || minCap > MAX_ARRAY_SIZE)
throw new OutOfMemoryError();
newCap = MAX_ARRAY_SIZE;
}
// 如果数组被更新了，就没有必要再分配新的空间了
if (newCap > oldCap && queue == array)
newArray = new Object[newCap];
} finally {
allocationSpinLock = 0;
}
}
// 其他线程正在占用allocationSpinLock，调用yield告诉线程调度
// 如果其他线程需要CPU，可以先拿去，我过会再执行，否则我继续执行。
if (newArray == null) // back off if another thread is allocating
Thread.yield();
// 因为要返回，再次获取主锁，而且后面可能要更新数组也需要主锁
lock.lock();
// 如果分配新空间成功，而且原先的队列没有被其他的线程更新过
// 就更新数组。这里不需要使用CAS，因为这个时候已经占用主锁了
if (newArray != null && queue == array) {
queue = newArray;
System.arraycopy(array, 0, newArray, 0, oldCap);
}再看取走元素的核心方法extract（poll方法也是使用这个方法从堆中拿走元素）
private E extract() {
E result;
int n = size - 1;
if (n < 0)
result = null;
else {
// 取走第一个元素
Object[] array = queue;
result = (E) array[0];
E x = (E) array[n];
array[n] = null;
Comparator<? super E> cmp = comparator;
if (cmp == null)
siftDownComparable(0, x, array, n);
else
siftDownUsingComparator(0, x, array, n, cmp);
size = n;
}
return result;
}这里提一下PriorityBlockingQueue的序列化。PriorityBlockingQueue内置了一个PriorityQueue对象，序列化会把元素转存到这个PriorityQueue中，然后再进行序列化。
反序列化时也是用PriorityQueue读取，然后再把元素转存回PriorityBlockingQueue自己的队列。
private PriorityQueue q;DelayQueue
DelayQueue有序存储Delayed类型或者子类型的对象，每当从队列中取走元素时，需要等待延迟耗完才会返回该对象。
所谓Delayed类型，因为需要比较，所以继承了Comparable接口：
public interface Delayed extends Comparable<Delayed> {
long getDelay(TimeUnit unit);
}其实Delayed对象的排序和延迟长短是无关的，因为Comparable的compare方法是用户自己实现的，DelayQueue只是保证返回对象的延迟已经耗尽。
DelayQueue需要排序存储Delayed类型的对象同时具备阻塞功能，但是阻塞的过程伴有延迟等待类型的阻塞，因此不能直接使用BlockingPriorityQueue来实现，而是用非阻塞的版本的PriorityQueue来实现排序存储。
private final PriorityQueue<E> q = new PriorityQueue<E>();因此DelayQueue需要自己实现阻塞的功能（需要一个Condition）：
private final Condition available = lock.newCondition(); public boolean offer(E e) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
q.offer(e);
// 如果原来队列为空，重置leader线程，通知available条件
if (q.peek() == e) {
leader = null;
available.signal();
}
return true;
} finally {
lock.unlock();
}
}顺便提一下，因为DelayQueue不限制长度，因此添加元素的时候不会因为队列已满产生阻塞，因此带有超时的offer方法的超时设置是不起作用的：
public boolean offer(E e, long timeout, TimeUnit unit) {
// 和不带timeout的offer方法一样
return offer(e);
}因为DelayQueue需要自己实现阻塞，因此关注的重点应该是两个带有阻塞的方法：没有超时的take方法和带有超时的poll方法。
普通poll方法很简单，如果延迟时间没有耗尽的话，直接返回null就可以了。
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
E first = q.peek();
if (first == null || first.getDelay(TimeUnit.NANOSECONDS) > 0)
return null;
else
return q.poll();
} finally {
lock.unlock();
}
}接下来看take和带timeout的poll方法，在看过DelayedWorkQueue之后这部分还是比较好理解的：
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
// 如果队列为空，需要等待available条件被通知
E first = q.peek();
if (first == null)
available.await();
else {
long delay = first.getDelay(TimeUnit.NANOSECONDS);
// 如果延迟时间已到，直接返回第一个元素
if (delay <= 0)
return q.poll();
// leader线程存在表示有其他线程在等待，那么当前线程肯定需要等待
else if (leader != null)
available.await();
else {
Thread thisThread = Thread.currentThread();
leader = thisThread;
// 如果没有leader线程，设置当前线程为leader线程
// 尝试等待直到延迟时间耗尽（可能提前返回，那么下次
// 循环会继续处理）
try {
available.awaitNanos(delay);
} finally {
// 如果leader线程还是当前线程，重置它用于下一次循环。
// 等待available条件时，锁可能被其他线程占用从而导致
// leader线程被改变，所以要检查
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
// 如果没有其他线程在等待，并且队列不为空，通知available条件
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}再来看带有timeout的poll方法，和DelayedWorkQueue非常相似：
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
long nanos = unit.toNanos(timeout);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
for (;;) {
E first = q.peek();
if (first == null) {
if (nanos <= 0)
return null;
else
// 尝试等待available条件，记录剩余的时间
nanos = available.awaitNanos(nanos);
} else {
long delay = first.getDelay(TimeUnit.NANOSECONDS);
if (delay <= 0)
return q.poll();
if (nanos <= 0)
return null;
// 当leader线程不为空时（此时delay>=nanos），等待的时间
// 似乎delay更合理，但是nanos也可以，因为排在当前线程前面的
// 其他线程返回时会唤醒available条件从而返回，
// 这里使用nanos和nonas<delay合并更加简单
if (nanos < delay || leader != null)
nanos = available.awaitNanos(nanos);
else {
Thread thisThread = Thread.currentThread();
leader = thisThread;
try {
long timeLeft = available.awaitNanos(delay);
// nanos需要更新
nanos -= delay - timeLeft;
} finally {
if (leader == thisThread)
leader = null;
}
}
}
}
} finally {
if (leader == null && q.peek() != null)
available.signal();
lock.unlock();
}
}前面理解了DelayedWorkQueue再来看DelayQueue就非常容易理解了。
