SynchronousQueue

SynchronousQueue是一种线程安全的同步移交的阻塞队列。SynchronousQueue比较特殊，它不像其他阻塞队列一样，有工作队列缓冲，它的每一个入队操作都必须对应其他线程的出队操作，没有数据容量。
SynchronousQueue有分为公平与非公平两个策略，公平策略是通过使用一个先进先出的队列来实现的，而非公平策略的话是通过后进先出的栈来保存的。

原理分析

SynchronousQueue主要是通过将对应的线程绑定在对应的队列节点中，通过LockSupport的unpark和park阻塞原语实现等待，先调用的线程需要调用LockSupport的park方法将调用方法的线程进行阻塞，直到另一个与之匹配的线程调用LockSupport.unpark来唤醒在该节点等待的线程。
在队列初始状态的时候，队列为null。当一个线程进行请求操作的时候，首先对队列进行判断，如果为null的空，将线程封装为对应的节点，加入到队列中等待，如果不为null的空，就对队列的第一个节点进行判断，是否匹配，如果匹配的话则进行交易，否则不匹配的话就加入到队列。

源码分析

构造函数

SynchronousQueue的构造函数可以传入一个boolean类型参数来决定使用公平策略还是非公平策略，默认使用非公平策略(false);
非公平策略是用TransferStack来实现的，它是SynchronousQueue的内部类，它继承了SynchronousQueue的Transferer内部抽象类
公平策略的话则是使用TransferQueue实现的，它跟TransferStack类似，不过一个后进先出，一个先进先出。

1
2
3

public SynchronousQueue(boolean fair) {
       transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
   }

常用函数

poll出队和offer入队方法都是调用transferer.transfer方法来实现的，它通过根据传入的元素是否为null来判断是生产者还是消费者，来判断是入队操作还是出队操作。具体的实现由子类TransferQueue和TransferStack来实现。

1
2
3
4
5
6

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
    E e = transferer.transfer(null, true, unit.toNanos(timeout));
    if (e != null || !Thread.interrupted())
        return e;
    throw new InterruptedException();
}

1
2
3
4
5
6
7
8
9

public boolean offer(E e, long timeout, TimeUnit unit)
    throws InterruptedException {
    if (e == null) throw new NullPointerException();
    if (transferer.transfer(e, true, unit.toNanos(timeout)) != null)
        return true;
    if (!Thread.interrupted())
        return false;
    throw new InterruptedException();
}

公平策略

成员变量

公平策略主要是通过一个先进先出的队列TransferQueue来保存节点。通过head和tail来指向队列的头部和尾部，cleanme是清除节点的时候需要用到的节点。

1
2
3
4
5
6
7
8
9
10

/** Head of queue */
transient volatile QNode head;
/** Tail of queue */
transient volatile QNode tail;
/**
 * Reference to a cancelled node that might not yet have been
 * unlinked from queue because it was the last inserted node
 * when it was cancelled.
 */
transient volatile QNode cleanMe;

next和tail节点都是SNode类型的,SNode中持有item字段表示要存储的元素数据，next指向队列的下一个节点，也是SNode类型的，waiter表示在该节点上等待的线程，isData用来标志该节点是由生产者还是消费者创建的(item!=null),生产者为true，消费者为false;

1
2
3
4

volatile QNode next;          // next node in queue
volatile Object item;         // CAS'ed to or from null
volatile Thread waiter;       // to control park/unpark
final boolean isData;

主要方法

• transfer
  transfer首先对当前队列进行判断，对当前队列为不为null的话或者当前线程节点与队尾节点的模式是否相等进行判断，如果为null或者模式相等的话，就将当前节点加入到队列中，即要保证队列中只能有一种模式，要么是消费者模式要么是生产者模式，因为一次入队必须有一次出队。
  入队操作主要分为两步，cas修改当前tail节点的next节点为当前要入队的节点，然后更新tail新节点。
  在多线程环境下，这两步可以在不同线程中完成。入队完成之后，需要使用LookSupport.park方法将当前线程阻塞，直到该线程被中断或者被唤醒或者设置了超时。
  如果节点被取消的话，就需要调用clean方法进行清除。
  如果节点匹配的话，因为是一个先进先出的队列，所以匹配的话肯定是发生在队列的头节点，只需要更新等待节点的item进行数据传输，然后调用LockSupport.unpark对等待线程进行唤醒。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

E transfer(E e, boolean timed, long nanos) {
    QNode s = null; // constructed/reused as needed
    boolean isData = (e != null);

    for (;;) {
        QNode t = tail;
        QNode h = head;
        if (t == null || h == null)         // saw uninitialized value
            continue;                       // spin

        if (h == t || t.isData == isData) { // empty or same-mode
            QNode tn = t.next;
            if (t != tail)                  // inconsistent read
                continue;
            if (tn != null) {               // lagging tail
                advanceTail(t, tn);
                continue;
            }
            if (timed && nanos <= 0)        // can't wait
                return null;
            if (s == null)
                s = new QNode(e, isData);
            if (!t.casNext(null, s))        // failed to link in
                continue;

            advanceTail(t, s);              // swing tail and wait
            Object x = awaitFulfill(s, e, timed, nanos);
            if (x == s) {                   // wait was cancelled
                clean(t, s);
                return null;
            }

            if (!s.isOffList()) {           // not already unlinked
                advanceHead(t, s);          // unlink if head
                if (x != null)              // and forget fields
                    s.item = s;
                s.waiter = null;
            }
            return (x != null) ? (E)x : e;

        } else {                            // complementary-mode
            QNode m = h.next;               // node to fulfill
            if (t != tail || m == null || h != head)
                continue;                   // inconsistent read

            Object x = m.item;
            if (isData == (x != null) ||    // m already fulfilled
                x == m ||                   // m cancelled
                !m.casItem(x, e)) {         // lost CAS
                advanceHead(h, m);          // dequeue and retry
                continue;
            }

            advanceHead(h, m);              // successfully fulfilled
            LockSupport.unpark(m.waiter);
            return (x != null) ? (E)x : e;
        }
    }
}

• awaitFulfill
  awaitFulfill阻塞线程或者自旋直到匹配，只有等待被取消、线程被中断或者匹配到对应的节点才会返回，如果等待被取消或者线程被中断都会当前节点，否则匹配到对应节点的话，就会返回匹配到的节点。
  如果节点不是队列的第一个节点，不需要自旋，减少空循环浪费cpu,如果是第一个节点的话，就根据是否设置超时来决定自旋次数。

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29

  Object awaitFulfill(QNode s, E e, boolean timed, long nanos) { /* Same idea as TransferStack.awaitFulfill */ 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); } }

• clean
  clean清除节点，不过不清除尾节点。如果需要清除尾节点的话，需要将cleanme赋值为尾节点，然后清除cleanme.

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50

  void clean(QNode pred, QNode s) { s.waiter = null; // forget thread /* * At any given time, exactly one node on list cannot be * deleted -- the last inserted node. To accommodate this, * if we cannot delete s, we save its predecessor as * "cleanMe", deleting the previously saved version * first. At least one of node s or the node previously * saved can always be deleted, so this always terminates. */ while (pred.next == s) { // Return early if already unlinked QNode h = head; QNode hn = h.next; // Absorb cancelled first node as head if (hn != null && hn.isCancelled()) { advanceHead(h, hn); continue; } QNode t = tail; // Ensure consistent read for tail if (t == h) return; QNode tn = t.next; if (t != tail) continue; if (tn != null) { advanceTail(t, tn); continue; } if (s != t) { // If not tail, try to unsplice QNode sn = s.next; if (sn == s || pred.casNext(s, sn)) return; } QNode dp = cleanMe; if (dp != null) { // Try unlinking previous cancelled node QNode d = dp.next; QNode dn; if (d == null || // d is gone or d == dp || // d is off list or !d.isCancelled() || // d not cancelled or (d != t && // d not tail and (dn = d.next) != null && // has successor dn != d && // that is on list dp.casNext(d, dn))) // d unspliced casCleanMe(dp, null); if (dp == pred) return; // s is already saved node } else if (casCleanMe(null, pred)) return; // Postpone cleaning s } }

非公平策略

SynchronousQueue的非公平策略主要是通过一个后进先出的栈TransferStack来实现的，通过持有一个head节点指向栈顶元素对栈的元素进行访问。

成员变量

1
2
3
4
5
6
7
8

/* Modes for SNodes, ORed together in node fields */
/** Node represents an unfulfilled consumer */
static final int REQUEST    = 0;
/** Node represents an unfulfilled producer */
static final int DATA       = 1;
/** Node is fulfilling another unfulfilled DATA or REQUEST */
static final int FULFILLING = 2;
volatile SNode head;

head字段是一个SNode类型的成员变量，SNode中next指向节点的下一个节点，也是SNode，match则是与当前节点匹配的节点，waiter则是在该节点上等待的线程，item则是节点要保存的数据，mode表示栈中节点的状态(REQUEST/DATA/FULFILLING|REQUEST/FULFILLING|DATA),next、match、waiter使用了volatile修饰，保证了其线程可见性和禁止重排序。

1
2
3
4
5

volatile SNode next;        // next node in stack
volatile SNode match;       // the node matched to this
volatile Thread waiter;     // to control park/unpark
Object item;                // data; or null for REQUESTs
int mode;

主要方法

• transfer
  TransferStack的transfer与TransferQueue不同，它不是将被匹配的节点出队列，然后将被匹配到节点压栈，然后将匹配的两个节点出栈。
  transfer首先对当前栈的状态进行判断。如果当前栈为null或者当前栈节点与栈顶节点的模式相等的话，就进行压栈操作。压栈的话，主要分为两步，将节点入栈，然后通过cas机制将head节点更新为当前节点。
  在完成压栈操作之后，需要通过LockSupport.park将当前线程阻塞，直到线程中断，或者节点被取消或者被匹配的节点唤醒。
  如果节点被取消的话，需要调用clean方法进行清理。
  由于是栈操作，后进先出，我们匹配到的节点肯定是栈顶的两个节点，我们需要原子性更新节点的匹配节点match和更新head节点。因为我们无法保证两步操作为原子性，需要将栈顶节点的mode设置为fulfiling，防止其他线程在栈顶进行匹配操作的时候对该节点进行其他操作，其他线程可以匹配操作进行辅助操作。

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56

  E transfer(E e, boolean timed, long nanos) { SNode s = null; // constructed/reused as needed int mode = (e == null) ? REQUEST : DATA; for (;;) { SNode h = head; if (h == null || h.mode == mode) { // empty or same-mode if (timed && nanos <= 0) { // can't wait if (h != null && h.isCancelled()) casHead(h, h.next); // pop cancelled node else return null; } else if (casHead(h, s = snode(s, e, h, mode))) { SNode m = awaitFulfill(s, timed, nanos); if (m == s) { // wait was cancelled clean(s); return null; } if ((h = head) != null && h.next == s) casHead(h, s.next); // help s's fulfiller return (E) ((mode == REQUEST) ? m.item : s.item); } } else if (!isFulfilling(h.mode)) { // try to fulfill if (h.isCancelled()) // already cancelled casHead(h, h.next); // pop and retry else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) { for (;;) { // loop until matched or waiters disappear SNode m = s.next; // m is s's match if (m == null) { // all waiters are gone casHead(s, null); // pop fulfill node s = null; // use new node next time break; // restart main loop } SNode mn = m.next; if (m.tryMatch(s)) { casHead(s, mn); // pop both s and m return (E) ((mode == REQUEST) ? m.item : s.item); } else // lost match s.casNext(m, mn); // help unlink } } } else { // help a fulfiller 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 } } } }

• awaitFulfill

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

  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; // establish waiter so can park next iter
        else if (!timed)
            LockSupport.park(this);
        else if (nanos > spinForTimeoutThreshold)
            LockSupport.parkNanos(this, nanos);
    }
}

• clean

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

void clean(SNode s) {
    s.item = null;   // forget item
    s.waiter = null; // forget thread


    SNode past = s.next;
    if (past != null && past.isCancelled())
        past = past.next;

    // Absorb cancelled nodes at head
    SNode p;
    while ((p = head) != null && p != past && p.isCancelled())
        casHead(p, p.next);

    // Unsplice embedded nodes
    while (p != null && p != past) {
        SNode n = p.next;
        if (n != null && n.isCancelled())
            p.casNext(n, n.next);
        else
            p = n;
    }
}

COMMENT AND SHARE