阻塞队列的首选之队列超集LinkedTransferQueue

落日映苍穹つ 2024-04-18 10:50 21阅读 0赞

### 文章目录 ###

*  前言
 *  队列的语义切换
 *   *  线程安全的队列
     *  Hand-off（SynchronousQueue语义）
     *  无界的阻塞队列
 *  LTQ原理分析
 *   *  假设阻塞特性
     *   *  1. 来一个线程存数据
         *  2. 又来一个线程存数据
         *  3. 来一个线程取数据
     *  总结

# 前言 #

这次讨论的议题是 JDK 1.7 新增的一个阻塞队列`LinkedTransferQueue`，那么它与[上一篇][Link 1]我们讨论的那3个基本的阻塞队列有什么不同呢？为什么JDK1.7的时候需要在JUC包中加入这个队列的实现呢？首先可以看看Doug Lea对增加LTQ（LinkedTransferQueue，下文都如此描述）这个队列的动机描述：

> 资料：http://cs.oswego.edu/pipermail/concurrency-interest/2009-February/005888.html
> 
> The simplest answer is that we realized that we should  
> have had this in the first place in Java5 rather than  
> some others, but didn’t know it. Better late than never.
> 
> The capabilities are a superset of those in
> 
>  *  ConcurrentLinkedQueue,
>  *  SynchronousQueue, in “fair” mode,
>  *  LinkedBlockingQueues that are not capacity bounded.
> 
> And better not only because you can now mix these capabilities,  
> but also because the implementation includes better techniques  
> we’ve discovered. So it seems like a good choice to place  
> in java.util.concurrent.

*  首先，doug表示这个队列的实现早该在Java5中出现在juc中，但是也没关系，迟来总比没来好。
 *  其次，此队列的能力是以下队列的超集
    
     *  ConcurrentLinkedQueue：非阻塞的高吞吐量的线程安全队列
     *  SynchronousQueue：阻塞队列，不存储元素，是hand-off的，吞吐量高
     *  LinkedBlockingQueues：基本的阻塞队列实现，可以存储元素
 *  最后，此队列不仅混合了以上能力，还附加了一些更好的技术（性能更好），综上所述，此队列似乎是juc中的一个很好的选择

这里再引入Pure Danger Tech这位小哥的博客的一段话

> 资料：https://puredanger.github.io/tech.puredanger.com/2009/02/28/java-7-transferqueue/
> 
> Joe Bowbeer helpfully [provided][] a link to a [paper by William Scherer, Doug Lea, and Michael Scott][paper by William Scherer_ Doug Lea_ and Michael Scott] that lays out the `LinkedTransferQueue` algorithms and performance tests showing their improvements over the existing Java 5 alternatives. `LinkedTransferQueue` outperforms `SynchronousQueue` by a factor of 3x in unfair mode and 14x in fair mode. Because `SynchronousQueue` is used as the heart of task handoff in something like `ThreadPoolExecutor`, this can result in similar kinds of performance improvements there. Given the importance of executors in concurrent programming, you start to see the importance of adding this new implementation.

*  这里表示，论文中表明LTQ的算法和性能测试表现了其改善了Java5中的队列选择，LTQ在非公平模式下优于SynchronousQueue 3个因子（不知道是不是3倍的意思，这里因子是3，应该是3倍吧？反正性能优），在公平模式下优于SynchronousQueue 14个因子

综上所述，LTQ性能更优，并且集合了（无界的）非阻塞队列（正常的线程安全队列FIFO）、阻塞队列（拥有阻塞特性的线程安全队列FIFO）、SIZE 0 （hand-off传输）的特性，在不失hand-off的特性的语义的同时可以灵活的在阻塞与非阻塞中切换，在需要线程安全的队列选择上应当优先被选择。

# 队列的语义切换 #

## 线程安全的队列 ##

TransferQueue<Integer> tlq = new LinkedTransferQueue<>();
    
    // ConcurrentLinkedQueue
    tlq.offer(1);
    tlq.put(2);
    
    // [1, 2]
    System.out.println(tlq);
    // 1
    System.out.println(tlq.poll());
    // 2
    System.out.println(tlq.poll());
    // null
    System.out.println(tlq.poll());

如果我们需要将其用作普通线程安全的队列，其没有阻塞的语义，有高吞吐量、伸缩性的保证

## Hand-off（SynchronousQueue语义） ##

// SynchronousQueue
    
    // blocking
    //tlq.transfer(1);
    
    // blocking
    //tlq.take()
    
    // 类似SynchronousQueue#offer
    // non-blocking,没人接收就丢弃了
    tlq.tryTransfer(1);
    // empty -> []
    System.out.println(tlq);

## 无界的阻塞队列 ##

// 无界的 LinkedBlockingQueue
    tlq.put(1);
    
    // 1
    System.out.println(tlq.poll());
    // size == 0 阻塞
    System.out.println(tlq.take());

# LTQ原理分析 #

那么其底层又是怎么做到同时可以有这么多的语义的呢？首先来看看几个API

public void transfer(E e) throws InterruptedException { 
      if (xfer(e, true, SYNC, 0) != null) { 
        Thread.interrupted(); // failure possible only due to interrupt
        throw new InterruptedException();
      }
    }
    public boolean tryTransfer(E e) { 
      return xfer(e, true, NOW, 0) == null;
    }
    public boolean add(E e) { 
      xfer(e, true, ASYNC, 0);
      return true;
    }
    // ...

其中使用了以下枚举来体现每个存取方法的语义

/* * Possible values for "how" argument in xfer method. */
    private static final int NOW   = 0; // for untimed poll, tryTransfer
    private static final int ASYNC = 1; // for offer, put, add
    private static final int SYNC  = 2; // for transfer, take
    private static final int TIMED = 3; // for timed poll, tryTransfer

可以看到，所有的存取api都调用了xfer这个方法，所以关键就在xfer方法中。这里我罗列了几个重要api调用xfer方法的参数列表：

<table> 
 <thead> 
  <tr> 
   <th>语义</th> 
   <th>队列存取方法</th> 
   <th>xfer方法参数列表</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>普通队列存取（non-blocking）</td> 
   <td>put/offer/add</td> 
   <td>xfer(e, true, ASYNC, 0)</td> 
  </tr> 
  <tr> 
   <td>非阻塞，队列空返回null</td> 
   <td>poll</td> 
   <td>xfer(null, false, NOW, 0)</td> 
  </tr> 
  <tr> 
   <td>hand-off（阻塞传球，队列不存放元素）</td> 
   <td>transfer</td> 
   <td>xfer(e, true, SYNC, 0)</td> 
  </tr> 
  <tr> 
   <td>类似于SynchronousQueue的offer</td> 
   <td>tryTransfer</td> 
   <td>xfer(e, true, TIMED, unit.toNanos(timeout))</td> 
  </tr> 
  <tr> 
   <td>若队列空，阻塞取</td> 
   <td>take</td> 
   <td>xfer(null, false, SYNC, 0)</td> 
  </tr> 
 </tbody> 
</table>

由于是无界的，所以添加元素不会阻塞，而阻塞取包含在take方法中，所以这里没有列出阻塞队列的语义。

## 假设阻塞特性 ##

在[上一篇文章][Link 1]中我们使用了假设法推敲SynchronousQueue的代码，这次我们依然这样做。在这一小节，我们假设在阻塞特性下模拟的场景。

### 1. 来一个线程存数据 ###

假设有一个线程阻塞存数据，首先调用了transfer(item)方法阻塞，直到另一个线程调用了take方法

// e, true, SYNC 0
    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
    
        // head此时为空，不进入分支
        for (Node h = head, p = h; p != null;) {  // find & match first node
          // ...
        }
    
        // 如果是立即的，可以看出这里就直接返回了
        // 这里是SYNC同步模式，进入分支
        if (how != NOW) {                  // No matches available
          // 进入分支
          if (s == null)
            // 构造新节点
            // Node.item = e,Node.haveData = true体现为存数据节点
            s = new Node(e, haveData);
          // 尝试将节点添加
          Node pred = tryAppend(s, haveData);
          // 从下面的tryAppend方法分析可以看出此时pred为s节点
          // 不进入此分支
          if (pred == null)
            continue retry;           // lost race vs opposite mode
    			// 由于不是ASYNC异步模式，所以会进入awaitMatch方法
          if (how != ASYNC)
            return awaitMatch(s, pred, e, (how == TIMED), nanos);
        }
        return e; // not waiting
      }
    }

这里进入tryAppend方法

// s = 刚刚构造的新节点，havaData = true
    private Node tryAppend(Node s, boolean haveData) { 
      // tail = null
      for (Node t = tail, p = t;;) {         // move p to last node and append
        Node n, u;                        // temps for reads of next & tail
        // 进入此分支
        if (p == null && (p = head) == null) { 
          // 将新节点设置为head
          if (casHead(null, s))
            // 返回
            return s;                 // initialize
        }
        //...
      }
    }

由于是SYNC模式，此时会进入awaitMatch方法

// s, s, e, timed = false, 0
    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;
        // s == s 不进入分支
        // 这里的意思是，如果节点s中的item变化了，表示有节点接收了此时存放的数据
        // 所以这就是退出for循环也就是有人来接收的标志
        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();
        }
        // 接下来的几次for循环都会进入这里，自旋
        // 由于阻塞挂起唤醒线程需要切换到内核态调用系统函数，有一定的开销
        // 为了不轻易挂起线程，所以会自旋
        else if (spins > 0) {              // spin
          --spins;
          if (randomYields.nextInt(CHAINED_SPINS) == 0)
            // 这里似乎会随机让出CPU时间片
            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);
        }
      }
    }

看到这里，应该已经发现了，大致的流程如下所示：

1.  将此时模式封装为Node，设置为head作为头节点
2.  自旋一会
3.  阻塞自己
4.  上述流程中，只要节点中的item改变了，就代表着有线程来接收，标志可以退出了

> 如果有看过我上一篇阻塞队列文章的读者，就会发现其实这里原理跟SynchronousQueue差不多，但论文中说的性能提升，笔者在这里猜想，是否是因为使用到了一些算法，类似于ConcurrentLinkedQueue的高吞吐算法，其他线程会协助迁移节点（这一点在下面会体现），因为除此之外，与SynchronousQueue并没有什么差别

### 2. 又来一个线程存数据 ###

整个存元素的流程走完了，这里假设此时又有一个线程进来存元素，事情又会变成怎么样呢？

// e, true, SYNC 0
    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
    
        // head此时不为空，循环直到 p指针为空
        for (Node h = head, p = h; p != null;) {  // find & match first node
          // isData = true
          boolean isData = p.isData;
          // 我们刚刚节点的item
          Object item = p.item;
          // 表示节点还没被处理，且 模式为存元素
          if (item != p && (item != null) == isData) {  // unmatched
            // 是一样的存模式，退出循环
            if (isData == haveData)   // can't match
              break;
            // ...
          }
          // ...
        }
    
        // 如果是立即的，可以看出这里就直接返回了
        // 这里是SYNC同步模式，进入分支
        if (how != NOW) {                  // No matches available
          // 进入分支
          if (s == null)
            // 构造新节点
            // Node.item = e,Node.haveData = true体现为存数据节点
            s = new Node(e, haveData);
          // 尝试将节点添加
          Node pred = tryAppend(s, haveData);
          // 从下面的tryAppend方法分析可以看出此时pred为s节点
          // 不进入此分支
          if (pred == null)
            continue retry;           // lost race vs opposite mode
    			// 由于不是ASYNC异步模式，所以会进入awaitMatch方法
          if (how != ASYNC)
            return awaitMatch(s, pred, e, (how == TIMED), nanos);
        }
        return e; // not waiting
      }
    }

这里的tryAppend方法的结果就和上面的不一样了

// haveData = true
    private Node tryAppend(Node s, boolean haveData) { 
      // tail = null
      for (Node t = tail, p = t;;) {         // move p to last node and append
        Node n, u;                        // temps for reads of next & tail
        // 这里head就不为null了，表示队列此时是有元素在的
        if (p == null && (p = head) == null) { 
          if (casHead(null, s))
            return s;                 // initialize
        }
        // 主要是判断此时的模式是否与p节点的匹配
        // 由于此时模式匹配，返回false不进入分支
        else if (p.cannotPrecede(haveData))
          return null;                  // lost race vs opposite mode
        // head节点的next=null
        else if ((n = p.next) != null)    // not last; keep traversing
          p = p != t && t != (u = tail) ? (t = u) : // stale tail
        (p != n) ? n : null;      // restart if off list
        // 将本节点设置为head的next节点
        // 然后继续循环
        else if (!p.casNext(null, s))
          p = p.next;                   // re-read on CAS failure
        // 第二次循环会走到这里
        else { 
          // 此时 t = null，进入分支
          if (p != t) {                  // update if slack now >= 2
            // 这里会执行casTail(t, s)，将此时节点设置为tail
            while ((tail != t || !casTail(t, s)) &&
                   (t = tail)   != null &&
                   (s = t.next) != null && // advance and retry
                   (s = s.next) != null && s != t);
          }
          return p;
        }
      }
    }

> 读者看到这里就会发现，此时如果有一个线程也是存元素，入队的时候并不会将其直接设置为tail，而是先设置为tail的next，在下一次循环才会设置为tail，其实这里的算法思想参考了ConcurrentLinkedQueue，这样会使得每个线程存取的吞吐量有一定的提升，感兴趣的读者可以了解一下这里的算法。

到这里我们发现一个规律，如果有线程频频不断的来存元素（是在hand-off特性下的假设），那么会先依次进入LTQ中的链表队列，然后自旋一会，如果还没人来处理自己，就会阻塞自身线程。

### 3. 来一个线程取数据 ###

那么，如果有一个线程来 take/poll ，事情又会是怎样呢？

// null, false, SYNC/NOW, 0
    // 其中 poll 为 NOW立即模式，take 为 SYNC同步模式，其实在这种情况下两者没什么分别
    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
    		// p = head，为第一个线程的Node对象
        // 当 p 为 null 时退出循环
        for (Node h = head, p = h; p != null;) {  // find & match first node
          // true
          boolean isData = p.isData;
          // 第一个线程想要传递的对象item
          Object item = p.item;
          // 进入分支
          if (item != p && (item != null) == isData) {  // unmatched
            // 因为此时是取模式，所以不会像以前那样退出循环
            if (isData == haveData)   // can't match
              break;
            // 将当时的null与第一个线程Node中的item交换
            // 第一个线程唤醒之后就会发现item改变，其就会退出了
            if (p.casItem(item, e)) {  // match
              // 此时p还是head，不进入循环
              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);
            }
          }
          // 如果上面的cas操作失败了，相当于取元素的时候被别的线程先取了
          // 那么拿到head的next节点，也就是第二个进来存的线程对应的Node
          // 继续上面操作，取元素唤醒线程
          Node n = p.next;
          p = (p != n) ? n : (h = head); // Use head if p offlist
        }
    
        // 如果走到这里，证明队列中已经没有元素了
        // 此时的take和poll才有区别
        // 如果是NOW，也就是poll，就会直接返回了，不阻塞
        if (how != NOW) {                  // No matches available
          // 如果是take，就会继续往下走
          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
      }
    }

到这里，已经足够看出规律了，不需要继续举例了。

*  队列为公平模式（因为要体现非阻塞队列的语义，其为FIFO），先进先出的模式，其体现在存元素时会往tail最后插入，取元素时会从head中取
 *  当模式为ASYNC或NOW，此时都不会阻塞，区别在于存的情况，如果是NOW模式，将不会存元素，直接返回了（类似SynchronousQueue的offer方法），如果是ASYNC模式，会像普通队列那样，往队列中塞元素进去
 *  当模式为SYNC，此时操作将都会阻塞
    
     *  存的情况：阻塞直到有线程来取
     *  取的情况：如果队列没有元素，阻塞直到有线程来存元素
 *  当模式为TIMED，此时操作类似阻塞（SYNC）的情况，但只会阻塞一段限定时间

## 总结 ##

综上所述，我们可以发现，LTQ使用了4种模式灵活的切换阻塞队列与非阻塞队列的语义，并且使用了更先进的算法改善了性能，但其大致实现原理跟SYnchronousQueue、ConcurrentLinkedQueue很像，内部维护一个链表结构，FIFO的存取元素，可以不阻塞的存取，也可以阻塞的存取（利用自旋和LockSupport\#park方法阻塞线程），保留了hand-off的特性的同时还能支持线程安全的非阻塞队列，实属线程安全队列之首选。其底层原理和算法可能不需要过多深究，但了解其API的使用和异同点还是很有必要的。

[Link 1]: https://blog.csdn.net/qq_41737716/article/details/100183193
[provided]: http://cs.oswego.edu/pipermail/concurrency-interest/2009-February/005886.html
[paper by William Scherer_ Doug Lea_ and Michael Scott]: http://www.cs.rice.edu/~wns1/papers/2006-PPoPP-SQ.pdf