LinkedTransferQueue

1. 理论

LinkedTransferQueue是一种同步队列，和SynchronousQueue相比增加了TransferQueue的抽象。以FIFO的方式传输数据，头节点的入队时间最早，尾节点入队时间最短。

TransferQueue是BlockingQueue的子接口。

public interface TransferQueue<E> extends BlockingQueue<E> {
    boolean tryTransfer(E e);
    void transfer(E e) throws InterruptedException;
    boolean tryTransfer(E e, long timeout, TimeUnit unit) throws InterruptedException;
    boolean hasWaitingConsumer();
    int getWaitingConsumerCount();
}

同步队列可以实现生产者等待消费者消费的队列，其中transfer()会等待消费者消费。

dual queue

放取元素使用同一个队列，队列中的节点具有两种模式，一种是数据节点，一种是非数据节点。

流程

LinkedTransferQueue采用的一种预占模式。 意思就是消费者线程取元素时，如果队列为空，那就生成一个节点（节点元素为null）入队，然后消费者线程park住，后面生产者线程入队时发现有一个元素为null的节点，生产者线程就不入队了，直接就将元素填充到该节点，唤醒该节点上park住线程，被唤醒的消费者线程拿货走人。这就是预占的意思：有就拿货走人，没有就占个位置等着，等到或超时。

• 放元素时先跟队列头节点对比，
  • 如果头节点是非数据节点，则进行匹配，
  • 如果头节点是数据节点，就生产一个数据节点放在队列尾端（入队）。
• 取元素时也是先跟队列头节点对比，
  • 如果头节点是数据节点，则进行匹配，
  • 如果头节点是非数据节点，就生产一个非数据节点放在队列尾端（入队）。

总之，类型不同则进行匹配并从对头出队，类型不同则从队尾入队。

队列中只会存储一种类型的节点

• 放数据生产者不会阻塞， 链表实现的无界队列
• 取数据是否阻塞看参数（NOW、ASYNC、SYNC、TIMED）

2. 实现

private E xfer(E e, boolean haveData, int how, long nanos) {
    if (haveData && (e == null))
        throw new NullPointerException();
    Node s = null;                        // the node to append, if needed

    retry:
    for (;;) {                            // restart on append race

        for (Node h = head, p = h; p != null;) { // find & match first node
            boolean isData = p.isData;
            Object item = p.item;
            if (item != p && (item != null) == isData) { // unmatched
                if (isData == haveData)   // can't match
                    break;
                if (p.casItem(item, e)) { // match
                    for (Node q = p; q != h;) {
                        Node n = q.next;  // update by 2 unless singleton
                        if (head == h && casHead(h, n == null ? q : n)) {
                            h.forgetNext();
                            break;
                        }                 // advance and retry
                        if ((h = head)   == null ||
                            (q = h.next) == null || !q.isMatched())
                            break;        // unless slack < 2
                    }
                    LockSupport.unpark(p.waiter);
                    return LinkedTransferQueue.<E>cast(item);
                }
            }
            Node n = p.next;
            p = (p != n) ? n : (h = head); // Use head if p offlist
        }

        if (how != NOW) {                 // No matches available
            if (s == null)
                s = new Node(e, haveData);
            Node pred = tryAppend(s, haveData);
            if (pred == null)
                continue retry;           // lost race vs opposite mode
            if (how != ASYNC)
                return awaitMatch(s, pred, e, (how == TIMED), nanos);
        }
        return e; // not waiting
    }
}

循环中分为两大部分，第一部分是匹配，第二部分是等待。

1. 节点如果取消，则item的值会为this，所以会先判断节点是否取消。
2. (item != null) == isData 判断头节点是否是未匹配过的。

if (item != p && (item != null) == isData) { // unmatched
    if (isData == haveData)   // can't match
        break;
}

3. 如果不能匹配，则跳出循环，进入入队等待处理。

if (p.casItem(item, e)) { // match
   ...
}

4. cas匹配，放元素则item是null->e, 取元素则item是e->null。
5. 匹配成功需要更新头节点，更新头节点要处考虑并发情况
   • 如果head还没变，则head更新为已经匹配过的next节点
   • 如果head已经变了，如果next是未匹配的，则已经被更新过了，直接跳出
6. 竞争失败，换下一个，下一个有问题则从head开始

p = (p != n) ? n : (h = head);

自旋和等待

private E awaitMatch(Node s, Node pred, E e, boolean timed, long nanos) {
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    Thread w = Thread.currentThread();
    int spins = -1; // initialized after first item and cancel checks
    ThreadLocalRandom randomYields = null; // bound if needed

    for (;;) {
        Object item = s.item;
        if (item != e) {                  // matched
            // assert item != s;
            s.forgetContents();           // avoid garbage
            return LinkedTransferQueue.<E>cast(item);
        }
        if ((w.isInterrupted() || (timed && nanos <= 0)) &&
                s.casItem(e, s)) {        // cancel
            unsplice(pred, s);
            return e;
        }

        if (spins < 0) {                  // establish spins at/near front
            if ((spins = spinsFor(pred, s.isData)) > 0)
                randomYields = ThreadLocalRandom.current();
        }
        else if (spins > 0) {             // spin
            --spins;
            if (randomYields.nextInt(CHAINED_SPINS) == 0)
                Thread.yield();           // occasionally yield
        }
        else if (s.waiter == null) {
            s.waiter = w;                 // request unpark then recheck
        }
        else if (timed) {
            nanos = deadline - System.nanoTime();
            if (nanos > 0L)
                LockSupport.parkNanos(this, nanos);
        }
        else {
            LockSupport.park(this);
        }
    }
}