简介
SynchronousQueue是一种特殊的阻塞队列，该队列没有容量。
【存数据线程】到达队列后，若发现没有【取数据线程】在此等待，则【存数据线程】便入队等待，直到有【取数据线程】来取数据，并释放【存数据线程】；
同理，【取数据线程】到达队列后，若发现没有【存数据线程】在此等待，则【取数据线程】便入队等待，直到有【存数据线程】来存数据，并释放【取数据线程】。
非公平模式，由伪栈实现，TransferStack
公平模式，由伪队列实现，TransferQueue
源码解析
Transferer
abstract static class Transferer<E> {
abstract E transfer(E e, boolean timed, long nanos); // 实现一个put或者take操作
}TransferStack
非公平模式
属性
static final int REQUEST = 0; // 未得到数据的消费者
static final int DATA = 1; // 未交出数据的生产者
static final int FULFILLING = 2; // 正在与另外一个消费者/生产者匹配
volatile SNode head; // stack的头结点
内部类SNode
属性
volatile SNode next; // 指向下一个结点
volatile SNode match; // 与当前结点匹配的结点
volatile Thread waiter; // 记录当前线程
Object item; // 数据，对于消费者，为空
int mode; // 模式，取值：REQUEST, DATA, FULFILLING
构造方法
SNode(Object item) {
this.item = item;
}关键方法
boolean casNext(SNode cmp, SNode val) { // CAS next属性，cmp -> val
return cmp == next && UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
}
boolean tryMatch(SNode s) { // 尝试匹配结点s和当前结点
if (match == null && UNSAFE.compareAndSwapObject(this, matchOffset, null, s)) { // 如果匹配成功
Thread w = waiter;
if (w != null) {
waiter = null;
LockSupport.unpark(w); // 唤醒阻塞的线程
}
return true;
}
return match == s;
}
void tryCancel() { // CAS match属性，null -> this
UNSAFE.compareAndSwapObject(this, matchOffset, null, this);
}
boolean isCancelled() { // 判断此结点是否已经被取消
return match == this;
}关键方法
static boolean isFulfilling(int m) { // 查看结点是否处于匹配模式
return (m & FULFILLING) != 0;
}
boolean casHead(SNode h, SNode nh) { // CAS head属性，h -> nh
return h == head && UNSAFE.compareAndSwapObject(this, headOffset, h, nh);
}
static SNode snode(SNode s, Object e, SNode next, int mode) { // 创建一个新的结点，或者重置s结点的相关域
if (s == null)
s = new SNode(e);
s.mode = mode;
s.next = next;
return s;
}
boolean shouldSpin(SNode s) { // 是否应该自旋等待
SNode h = head;
return (h == s || h == null || isFulfilling(h.mode));
}transfer()
基本的算法是在一个无限循环中，每次执行下面三种情况的其中一种：
- 如果当前栈为空，或者已经包含与当前结点模式相同的结点，尝试入栈，并一直等待，直到遇到与之匹配(模式互补)的结点前来将其唤醒，并返回结果：a. 如果被取消，则返回null; b.如果当前结点是消费者，则返回匹配结点的数据；c. 如果当前结点是生产者，则返回当前结点的数据。
- 如果栈中包含与当前结点模式互补的结点，则设置当前结点的模式为FULFILLING，并尝试入栈，和对应的结点互相匹配，完成后，双双出栈，并返回生产者的数据。
- 如果栈顶结点的模式是FULFILLING，说明此刻有结点正在配对，当前线程帮助它们配对和弹出栈，然后在处理自己的事情，继续循环执行相应的操作。
E transfer(E 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))) { // 构建结点入栈
SNode m = awaitFulfill(s, timed, nanos); // 在此等待匹配结点唤醒自己
if (m == s) { // 已取消
clean(s); // 清理工作
return null;
}
if ((h = head) != null && h.next == s)
casHead(h, s.next); // 帮助配对成功的结点出栈
return (E) ((mode == REQUEST) ? m.item : s.item); // 返回生产者的数据
}
} else if (!isFulfilling(h.mode)) { // 尝试匹配结点
if (h.isCancelled()) // 已经被取消
casHead(h, h.next); // 更新head
else if (casHead(h, s = snode(s, e, h, FULFILLING | mode))) { // 设置为FULFILLING模式入栈
for (;;) { // 寻找匹配者
SNode m = s.next; // m是s的匹配者
if (m == null) { // 为空
casHead(s, null); // 弹出已经配对的结点
s = null; // 下一次重新构造结点
break; // 重新开始
}
SNode mn = m.next;
if (m.tryMatch(s)) {
casHead(s, mn); // 弹出s和m
return (E) ((mode == REQUEST) ? m.item : s.item); // 返回生产者的数据
} else // 失败
s.casNext(m, mn); // 剔除m
}
}
} else {
SNode m = h.next; // m是h的匹配者
if (m == null) // 为空
casHead(h, null); // 弹出已经配对的结点
else {
SNode mn = m.next;
if (m.tryMatch(h)) // 帮助完成匹配
casHead(h, mn); // 弹出h和m
else // 失败
h.casNext(m, mn); // 剔除m
}
}
}
}awaitFulfill()
在线程阻塞之前，设置到结点的waiter域，并且检查一次线程线程的中断状态，若中断则取消。如果执行结点处于栈顶，阻塞之前会自旋一会儿，说不定马上就有结点来匹配，这样就不用阻塞了。主循环检查顺序: 中断优先于正常返回，正常返回优先于超时。
SNode awaitFulfill(SNode s, boolean timed, long nanos) {
final long deadline = timed ? System.nanoTime() + nanos : 0L; // 超时时间点
Thread w = Thread.currentThread(); // 当前线程
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) { // 检查超时
nanos = deadline - System.nanoTime();
if (nanos <= 0L) { // 若超时，取消此结点
s.tryCancel();
continue;
}
}
if (spins > 0) // 自旋
spins = shouldSpin(s) ? (spins - 1) : 0;
else if (s.waiter == null)
s.waiter = w; // 记录线程
else if (!timed)
LockSupport.park(this); // 阻塞线程
else if (nanos > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanos); // 带超时的阻塞
}
}clean()
清理分3步
- 清理s结点，并判断s的next结点past，如果past也取消了，则跳过此结点，使得past变量指向下一个结点，到此为止，此为清理工作的最大深度
- 从head结点开始，依次跳过已经取消的结点，直到遇到未取消的结点（或者遇到past结点，或为空），重新设置head结点为p结点
- 从p到past结点，清理掉所有已经取消的结点
void clean(SNode s) {
s.item = null; // 清理s结点
s.waiter = null;
SNode past = s.next; // s的下一个结点past
if (past != null && past.isCancelled()) // 如果past也取消了，则直接跳到past的下一个结点
past = past.next;
SNode p;
while ((p = head) != null && p != past && p.isCancelled()) // 从head结点开始，遍历清理已经取消的结点，直到遇到没有被取消的结点，并设置为新的head结点
casHead(p, p.next);
while (p != null && p != past) { // 从p结点到past结点(但不包括past), 遍历清理所有已经取消的结点
SNode n = p.next;
if (n != null && n.isCancelled())
p.casNext(n, n.next);
else
p = n;
}
}TransferQueue
公平模式
内部类QNode
属性
volatile QNode next; // 指向下一个结点
volatile Object item; // 存放数据，isData为false时，该节点为null, 为true时，匹配后，该节点会置为null
volatile Thread waiter; // 控制线程的park/unpark
final boolean isData; // 表示该结点是存数据还是取数据
关键方法
QNode(Object item, boolean isData) { // 构造方法
this.item = item;
this.isData = isData;
}
boolean casNext(QNode cmp, QNode val) { // CAS next域 cmp -> val
return next == cmp && UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
}
boolean casItem(Object cmp, Object val) { // CAS offset域 cmp -> val
return item == cmp && UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
}
void tryCancel(Object cmp) { // 取消结点
UNSAFE.compareAndSwapObject(this, itemOffset, cmp, this);
}
boolean isCancelled() { // 检查是否已取消
return item == this;
}
boolean isOffList() { // 检查结点是否离开了队列
return next == this;
}属性
transient volatile QNode head; // 头结点
transient volatile QNode tail; // 尾结点
transient volatile QNode cleanMe; // 当一个结点被标记取消时，恰巧又是最后也给结点，那么将cleanMe作为该结点的predecessor
关键方法
TransferQueue() { // 构造方法
QNode h = new QNode(null, false); // 初始化 dummy node.
head = h;
tail = h;
}
void advanceHead(QNode h, QNode nh) { // CAS head域，h -> nh
if (h == head && UNSAFE.compareAndSwapObject(this, headOffset, h, nh))
h.next = h;
}
void advanceTail(QNode t, QNode nt) { // CAS tail域，t -> nt
if (tail == t)
UNSAFE.compareAndSwapObject(this, tailOffset, t, nt);
}
boolean casCleanMe(QNode cmp, QNode val) { // CAS cleanMe域，cmp -> val
return cleanMe == cmp && UNSAFE.compareAndSwapObject(this, cleanMeOffset, cmp, val);
}transfer()
基本的算法是在一个无限循环中，每次执行下面两种情况的其中一种：
- 如果当前队列为空，或者已经与当前结点模式相同的结点，尝试入队，并一直等待，直到遇到与之匹配(模式互补)的结点前来将其唤醒，并返回匹配结点的数据。
- 如果队列中包含与当前结点模式互补的结点，则尝试和对应的结点互相匹配，完成后，将等待结点出队，并返回匹配结点的数据。
- 在每个动作里面，都会检测并帮助其他线程来完成节点推进。
E transfer(E e, boolean timed, long nanos) {
QNode s = null;
boolean isData = (e != null);
for (;;) {
QNode t = tail;
QNode h = head;
if (t == null || h == null) // 当前线程看到未初始化的头尾结点
continue; // 自旋
if (h == t || t.isData == isData) { // 队列为空，或者包含相同模式的结点
QNode tn = t.next;
if (t != tail) // 过期数据
continue;
if (tn != null) { // 别的线程添加了新的结点，帮助更新tail域
advanceTail(t, tn);
continue;
}
if (timed && nanos <= 0) // 超时
return null;
if (s == null)
s = new QNode(e, isData); // 构造结点
if (!t.casNext(null, s)) // 连接失败
continue;
advanceTail(t, s); // 设置s为tail结点
Object x = awaitFulfill(s, e, timed, nanos); // 等待匹配
if (x == s) { // 如果取消，清理结点
clean(t, s);
return null;
}
if (!s.isOffList()) { // 如果s未离队
advanceHead(t, s); // 设置s为head结点
if (x != null)
s.item = s;
s.waiter = null;
}
return (x != null) ? (E) x : e;
} else { // 互补模式
QNode m = h.next;
if (t != tail || m == null || h != head)
continue; // 读取的是过期的值，继续循环
Object x = m.item;
if (isData == (x != null) || // m已经被匹配了
x == m || // m被取消
!m.casItem(x, e)) { // CAS失败
advanceHead(h, m); // h出队，m设置为head结点，重来
continue;
}
advanceHead(h, m); // 成功，推进头节点
LockSupport.unpark(m.waiter); // 唤醒等到线程
return (x != null) ? (E) x : e;
}
}
}awaitFulfill()
Object awaitFulfill(QNode s, E e, boolean timed, long nanos) {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
Thread w = Thread.currentThread();
int spins = ((head.next == s) ? (timed ? maxTimedSpins : maxUntimedSpins) : 0);
for (;;) {
if (w.isInterrupted())
s.tryCancel(e);
Object x = s.item;
if (x != e)
return x;
if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
s.tryCancel(e);
continue;
}
}
if (spins > 0)
--spins;
else if (s.waiter == null)
s.waiter = w;
else if (!timed)
LockSupport.park(this);
else if (nanos > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanos);
}
}同TransferStack.
clean()
在任意时间点，只有最后入队的结点不能立即删除，因为考虑到无所并发，线程争用下沉到CPU指令级别（CAS），最后入队的结点同时会有CAS Tail的动作，所以不能同一时间点，对同一个元素有多个CAS操作，因此，如果是最后入队的结点，可以将删除操作滞后。先将cleanMe结点的next域指向改结点，等到有新的结点入队时，再考虑删除上一版本的结点，此时，已满足条件。
void clean(QNode pred, QNode s) {
s.waiter = null; // 置空waiter域
while (pred.next == s) {
QNode h = head;
QNode hn = h.next;
if (hn != null && hn.isCancelled()) {
advanceHead(h, hn); // 推进head结点
continue;
}
QNode t = tail;
if (t == h)
return;
QNode tn = t.next;
if (t != tail)
continue;
if (tn != null) {
advanceTail(t, tn);
continue;
}
if (s != t) { // 如果s不是尾结点，直接将其删除
QNode sn = s.next;
if (sn == s || pred.casNext(s, sn))
return;
}
QNode dp = cleanMe;
if (dp != null) { // 尝试删除前一版本取消的结点，借助cleanMe结点
QNode d = dp.next;
QNode dn;
if (d == null || // d已经被删除
d == dp || // d已经出
!d.isCancelled() || // d没被取消
(d != t && // d not tail and
(dn = d.next) != null && // 有后继结点
dn != d && dp.casNext(d, dn))) // 删除d
casCleanMe(dp, null);
if (dp == pred)
return;
} else if (casCleanMe(null, pred))
return;
}
}行文至此结束。
尊重他人的劳动，转载请注明出处：http://www.cnblogs.com/aniao/p/aniao_sq.html