源码阅读(38)：Java中线程安全的QueueDeque结构——LinkedTransferQueue

Posted 2022-12-02 说好不能打脸

1、LinkedTransferQueue概述

LinkedTransferQueue是从JDK 1.7+版本开始提供的一个无界阻塞式队列，它是Java容器框架中一种比较特殊的阻塞式队列，特殊性体现在它实现的TransferQueue接口。后者的特点是可定义一种数据对象消费者和生产者的配对交换方式，保证了生产者线程和消费者线程的配对处理（注意，不是数据配对而是线程配对），这样做的好处是，可以使用CAS原理进行LinkedTransferQueue队列集合的线程安全性控制，而不是使用AQS原理，在大多数高并发场景下，基于CAS工作的线程安全的数据结构其性能由于基于AQS工作的线程安全的数据结构。

从JDK 9+版本开始，LinkedTransferQueue内部的实现机制做了较大的调整，主要是使用变量句柄替代了
sun.misc.Unsafe工具类，并优化了内部结构的实现性能。本章节对于LinkedTransferQueue队列集合的讲解，将直接基于JDK 9+的源代码版本。下图是LinkedTransferQueue队列集合的主要继承体系：

LinkedTransferQueue队列集合的使用场景同样是基于生产者和消费者的多线程场景，在使用层面上该队列集合和我们已经介绍过的ArrayBlockingQueue、LinkedBlockingQueue等队列类似，如下所示（简单的消费者和生产者示例）：

// ......
public static void main(String[] args) 
  // 本文建议使用线程池管理线程，而不是直接创建Thread对象
  ThreadPoolExecutor serviceExecutor = new ThreadPoolExecutor(10,10,1000,TimeUnit.SECONDS,new LinkedBlockingQueue<>());
  // 这是一个LinkedTransferQueue队列
  LinkedTransferQueue<String> queue = new LinkedTransferQueue<>();
  // 多个消费者和多个生产者
  serviceExecutor.submit(new Producer(queue));
  serviceExecutor.submit(new Producer(queue));
  serviceExecutor.submit(new Consumer(queue));
  serviceExecutor.submit(new Consumer(queue));


// 消费者
public static class Consumer implements Runnable 
  private TransferQueue<String> queue;
  
  public Consumer(TransferQueue<String> queue) 
    this.queue = queue;
  
  
  @Override
  public void run() 
    int count = 0;
    while(count++ < Integer.MAX_VALUE) 
      try 
        String value = this.queue.take();
        System.out.println(value);
       catch (InterruptedException e) 
        e.printStackTrace(System.out);
      
    
  


// 生产者
public static class Producer implements Runnable 
  // 生产者生产的数据，将放入该队列
  private TransferQueue<String> queue;
  public Producer(TransferQueue<String> queue) 
    this.queue = queue;
  

  @Override
  public void run() 
    String uuid = UUID.randomUUID().toString();
    int count = 0;
    while(count++ < Integer.MAX_VALUE) 
      // 通过transfer的方式进行添加
      try 
        this.queue.transfer(uuid);
       catch (InterruptedException e) 
        e.printStackTrace(System.out);
      
    
  

请注意，和之前利用阻塞队列实现的生产者/消费者进行比较，最有特点的就是生产者可以采用transfer将数据对象添加到LinkedTransferQueue队列中，如果这个数据对象暂时没有消费者线程取出进行处理，则当前生成者会进入阻塞状态。而消费者依然可以使用类似的方式（take方法）试图从队列中取出数据对象，如果没有数据对象可以取出，则消费者进入阻塞状态。

2、LinkedTransferQueue核心结构

2.1、LinkedTransferQueue中的单向链表

LinkedTransferQueue内部采用一个无界单向链表来连续记录需要从LinkedTransferQueue队列取数据对象的消费者请求，或者向LinkedTransferQueue队列添加数据对象的生产者请求。如下图所示：

请注意，head引用位置不一定绝对在单向链表的首个Node对象上，tail引用位置也不一定绝对在单向链表的最后一个Node对象上（后续文章会重点讲到这个问题）。如上图所示，单向链表的每一个节点都是一个LinkedTransferQueue.Node类的实例，每一个Node类中不一定有数据，也不一定有消费者线程或者生产者线程的对象引用。Node类定义的代码片段如下所示：

public class LinkedTransferQueue<E> extends AbstractQueue<E>
    implements TransferQueue<E>, java.io.Serializable 
    // ......
    // 该属性记录单向链表的第一个有效Node节点位置
    // 不一定是单向链表的头节点
    transient volatile Node head;
    // 该属性记录单向链表的最新知晓的在多线程操作场景下正确入队的最后一个Node节点位置
    // 不一定是单向链表的尾节点
    private transient volatile Node tail;
    
	// ......
	static final class Node 
	  // ......
	  // 表示该Node节点是否存储了数据
	  // 如果没有存储数据，那么说明这个节点记录的是一个消费者请求
	  // 否则就是记录的消费者请求
	  // false if this is a request node
	  final boolean isData;   
	  // 如果该节点记录了数据，则数据对象通过该属性被引用
	  // initially non-null if isData; CASed to match
	  volatile Object item;
	  // 该属性指向当前节点的下一个节点，以便构成一个链表
	  volatile Node next;
	  // 造成当前节点所记录的请求，对应的线程
	  // null when not waiting for a match
	  volatile Thread waiter; 
	  // ......
	
	// ......

注意，正常情况下Node对象中的isData属性和item属性所描述的结论应该是一致的，也就是说当isData属性值为false时，item属性值应该为null；当isData属性值为true时，item属性值应该不为null。但是在单向链表工作过程中，经常会出现isData属性和item属性描述结论不一致的情况，这是正常情况，这种节点称为“虚”节点（或无效节点）。

当一个消费者线程向LinkedTransferQueue请求取得数据时，如果LinkedTransferQueue队列中没有任何可以取出的数据，那么将为这个取数请求创建一个节点，这个节点将会被添加到当前链表的末尾，同时这个取数请求的线程将进入阻塞状态（parking_waiting）。当然根据调用方法不同，取数请求的线程也有可能不会被阻塞，但是这于核心原理并没有本质影响。

随后，当一个生产者线程向LinkedTransferQueue请求添加数据时，会从单向链表的head节点开始匹配，如果发现某个节点是一个取数请求任务类型的节点（即是这个节点的isData为false，item == null），那么则将数据添加到这个节点上，并通知取数请求的线程，解除阻塞状态。如下图所示：

注意，head引用位置上的节点，可能是一个“虚”节点，这时处理过程会继续向后寻找，直到匹配到第一个有效的任务节点。

并且，以上处理规则是可以反过来的： 当一个生产者线程向LinkedTransferQueue请求添加数据时，如果LinkedTransferQueue队列中没有任何等待取数的请求节点，那么将为这个生产者的添加请求创建一个节点，这个节点将会被添加到当前链表的末尾，同时这个添加请求的线程将视调用情况进入阻塞状态（parking_waiting）…… 如下图所示：

请注意，以上处理逻辑隐含了一个潜在规则，就是LinkedTransferQueue内部链表上的有效节点，要么全部都是由取数请求创建的节点，其isData为false，item属性为null；要么就全部都是由存储请求创建的节点，其isData为true，item属性不为null。这就是为什么与之对应的消费者或者生产者，都只需要由head开始找到第一个有效节点判定是否可以存储/添加数据，而不需要对这个链表上的所有节点性质进行判定的原因。

经过对LinkedTransferQueue处理逻辑的初步分析，我们可以得出以下几个显而易见的结论：

• 基于LinkedTransferQueue工作的生产者和消费者，其添加/取出数据的理论时间复杂度为O（1）

• LinkedTransferQueue队里集合采用CAS思想而非AQS思想保证自身的线程安全性。

• 要保证LinkedTransferQueue的线程安全性，本质上就是在多线程（多生产者多消费者）的场景下维护这个单向链表。要知道实际的工作场景下，有多个生产者线程和消费者线程同时发起对这个单向链表的操作，如下图所示：
  

2.2、VarHandle变量句柄

从JDK 9+开始，Java提供了一个VarHandle变量句柄功能，用以推荐给程序员支持自行编写的CAS思路的程序逻辑落地。在这之前，JDK内部实现CAS思想时，大量使用了UnSafe工具类。后者这是一个直连Hotspot VM后门的工具类（在前文中我们已经介绍过《多任务处理（17）——Java中的锁（Unsafe基础）》），如果不修改虚拟机的安全性规则，编译器不允许程序员直接使用这个类，并且一般情况下程序员也需要使用“反射”的方式获得UnSafe工具类的操作对象。

2.2.1、利用java.util.concurrent.atomic包完成原子操作

Unsafe工具类中基本都是基于JIN直接对内存进行的操作，或者说直接通过C对硬件进行的操作。Java语言的各个版本设计，都不希望程序员冲破这个封装，所以才会对UnSafe工具类的使用做诸多限制。在JDK 9 之前技术人员要自行完成CAS思想的落地实际上除了使用UnSafe工具类外，实际上就是使用java.util.concurrent.atomic原子操作包提供的各种工具类——例如技术人员要利用CAS思想在多线程场景下，完成某个对象属性的线程安全性的赋值操作，那么通常还有以下这些选择（可选择的方式确实算不上丰富）：

• 使用java.util.concurrent.atomic包下的某个原子操作工具，辅助完成CAS判定过程，例如使用AtomicBoolean模拟抢占与自旋操作。基本思路是：多线程场景下，成功将AtomicBoolean对象从false赋值为true的线程，视为抢占成功的线程，可以进行对象属性的赋值操作，其它线程进入“自旋”。代码格式通常如下：

// 这是多个线程抢着要使用的资源
private XXXXX filed1;
// 这是辅助完成cas判定的原子操作工具
private AtomicBoolean isOk = new AtomicBoolean(false);

// .....
for(;;) 
  // 如果条件成立，说明当前线程抢占到了资源操作权
  // 此方法的意思是，如果当前isOk的布尔值为false，则将isOk的布尔值设置为true，并返回原子操作成功的信息。
  // 否则返回false，代表原子操作失败
  if(isOk.compareAndSet(false , true)) 
    try 
      // 这里进行filed1赋值操作和业务逻辑处理
      // .........
      return;
     finally 
      // 无论处理是否成功，业务操作后，都要将isOk设置为false
      // 以便其它还在自旋的线程能够继续利用CAS的思想进行竞争
      isOk.set(false);
    
  

// .....

这种处理选择的问题在于所需的代码段落比较多，实际上AtomicBoolean内部就是UnSafe工具类的封装。

• 另外还有一种辅助完成CAS的方式，是借助java.util.concurrent.atomic包中的AtomicReference对象，完成被引用的对象更新。

之前的文章已经介绍过，AtomicReference和AtomicBoolean、AtomicInteger等原子操作类工作原理类似，不同的是前者可以对更泛化的对象引用进行原子操作，而不像后者那样，只是基于特定的boolean、int基础类型进行操作。 基本的使用模式可以如下所示：

public class ...... 
  // 这是多个线程抢着要使用的资源
  private YYYY filed1 = new YYYY();
  
  // ......
  // 建立filed1属性的原子操作引用
  AtomicReference atomicReference = new AtomicReference(filed1);
  // .....

在JDK 9之前的版本，AtomicReference内部还是基于UnSafe工具类的封装，但是在JDK9 +的版本开始，AtomicReference内部的封装改为了VarHandle变量句柄。

• 还可以借助java.util.concurrent.atomic包中诸如AtomicIntegerFieldUpdater这样的字段原子更新工具或者使用支持更泛化类型的AtomicReferenceFieldUpdater原则更新工具。

这个使用方式在之前的文章中也进行了介绍，这里就不再赘述，基础思想也是对UnSafe工具类的封装。由此可见如果技术人员遵循Java官方要求，。所以。通过以上这些介绍，读者应该可以更加明确的知道为什么java.util.concurrent.atomic包叫做原子操作包了。

2.2.2、利用VarHandle变量句柄完成工作

从JDK 9+开始，如果开发人员遵循Java推荐规范不直接使用UnSafe工具类来落地自己的CAS实现，那么除了使用java.util.concurrent.atomic原子操作包外，还可以使用新提供的VarHandle变量句柄。VarHandle变量句柄不止是Java推荐开发人员使用的落地CAS的方式，JDK内部也在逐渐使用VarHandle变量句柄的方式，替换掉以前大规模直接使用的UnSafe工具类。

例如AtomicReference的内部、前文介绍的PriorityBlockingQueue队列集合；再例如，本文正在介绍的LinkedTransferQueue队列，其源代码中也充分利用了VarHandle变量句柄完成CAS过程，其中和VarHandle直接有关的定义如下：

// ......
public class LinkedTransferQueue<E> extends AbstractQueue<E>
    implements TransferQueue<E>, java.io.Serializable 
  // ......
  // VarHandle mechanics
  // 为以下属性建立操作句柄，以简化CAS的编码过程。
  private static final VarHandle HEAD;
  private static final VarHandle TAIL;
  private static final VarHandle SWEEPVOTES;-
  static final VarHandle ITEM;
  static final VarHandle NEXT;
  static final VarHandle WAITER;
  static 
    try 
      MethodHandles.Lookup l = MethodHandles.lookup();
      // 建立LinkedTransferQueue类中，head属性的操作句柄
      HEAD = l.findVarHandle(LinkedTransferQueue.class, "head", Node.class);
      // tail属性的操作句柄
      TAIL = l.findVarHandle(LinkedTransferQueue.class, "tail", Node.class);
      // sweepVotes属性的操作句柄
      SWEEPVOTES = l.findVarHandle(LinkedTransferQueue.class, "sweepVotes", int.class);
      // LinkedTransferQueue.Node类中，item属性的操作句柄
      ITEM = l.findVarHandle(Node.class, "item", Object.class);
      // LinkedTransferQueue.Node类中，next属性的操作句柄 
      NEXT = l.findVarHandle(Node.class, "next", Node.class);
      // LinkedTransferQueue.Node类中，waiter属性的操作句柄
      WAITER = l.findVarHandle(Node.class, "waiter", Thread.class);
     catch (ReflectiveOperationException e) 
      throw new ExceptionInInitializerError(e);
    
    // ......
  
  // ......

// ......

LinkedTransferQueue队列主要通过CAS原理保证多线程场景下内部单向链表结构的数据准确性，而进行CAS思想落地的主要方式就是依靠VarHandle变量句柄。

========（接下文）