生产-消费模型之"阻塞队列"的源码分析

比眉伴天荒 2024-04-17 21:24 45阅读 0赞

### 文章目录 ###

*  前言
 *  阻塞队列API
 *   *  存放元素
     *   *  boolean add(E e)
         *  boolean offer(E e)
         *  boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException;
         *  void put(E e) throws InterruptedException
     *  获取元素
     *   *  E take() throws InterruptedException
         *  E poll(long timeout, TimeUnit unit) throws InterruptedException
     *  小结
 *  ArrayBlockingQueue
 *   *  offer（无时限）
     *  offer（有时限）
     *  put(E e)
     *  小结
 *  LinkedBlockingQueue
 *   *  小结
 *  SynchronousQueue
 *   *  有一个线程调用put方法
     *  另一个线程调用take方法
     *  小结

# 前言 #

通常来说，Queue队列的特性一般为FIFO（先进先出），在队列的特性上在增加一个阻塞的特性，这就是BlockingQueue，这很像一个生产者-消费者模式，生产者负责将某些元素放入队列，消费者负责从队列中取元素，阻塞-通知的特性让这种模式更加高效，例如有元素我就会唤醒阻塞的线程，而不是一直去主动轮询是否有任务，在线程池中也利用了阻塞队列的特性，实现了线程任务的提交和获取执行，所以配置一个合适的线程池之前，你需要了解阻塞队列的实现和使用。

本篇文章的议题围绕BlockingQueue几个常用的阻塞队列实现类：

*  SynchronousQueue
 *  ArrayBlockingQueue
 *  LinkedBlockingQueue

但阻塞队列中有些方法是会一直阻塞，有些方法又不会，每个方法都有其特有的场景和作用，在这里会介绍其源码，分析其原理，让读者更好的使用和了解阻塞队列的特性。

不同阻塞队列实现类，内部的数据结构和行为都存在一定的不同，所以本篇文章的维度将在不同阻塞队列的实现和实现对应的一系列存取方法来展开。

# 阻塞队列API #

那么阻塞队列都有哪些方法可供用户使用呢？直接来看J.U.C中的BlockingQueue这个接口

## 存放元素 ##

### boolean add(E e) ###

> Inserts the specified element into this queue if it is possible to do so immediately without violating capacity restrictions, returning \{@code true\} upon success and throwing an \{@code IllegalStateException\} if no space is currently available.  
> When using a capacity-restricted queue, it is generally preferable to use \{@link \#offer(Object) offer\}.

此方法上的注释说明了此方法的特性：

*  会立即返回
    
     *  返回true代表添加成功
     *  抛出IllegalStateException异常代表容量限制
 *  如果是一个有容量限制的队列，一般来说更**偏向使用offer方法**去存放元素

### boolean offer(E e) ###

> Inserts the specified element into this queue if it is possible to do so immediately without violating capacity restrictions, returning \{@code true\} upon success and \{@code false\} if no space is currently available. When using a capacity-restricted queue, this method is generally preferable to \{@link \#add\}, which can fail to insert an element only by throwing an exception.

*  立即返回
    
     *  返回ture代表添加成功
     *  返回false代表添加失败，当前没有可用空间
 *  在有容量限制的队列中，通常这个方法比add要好

### boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException; ###

> Inserts the specified element into this queue, waiting up to the specified wait time if necessary for space to become available.

*  会阻塞一段用户自定的timeout时间，为阻塞一段时间版的offer
    
     *  因为会阻塞，所以有可能抛出InterruptedException中断的异常

### void put(E e) throws InterruptedException ###

> Inserts the specified element into this queue, waiting if necessary for space to become available.

*  如果队列没有空间存放元素会一直阻塞
    
     *  若方法返回，代表添加成功
     *  因为会阻塞，所以此方法有可能抛出InterruptedException中断的异常

## 获取元素 ##

### E take() throws InterruptedException ###

> Retrieves and removes the head of this queue, waiting if necessary until an element becomes available.

*  取出然后删除队列的头元素
 *  如果没有元素在队列，方法会一直阻塞
    
     *  同样需要处理中断异常

### E poll(long timeout, TimeUnit unit) throws InterruptedException ###

> Retrieves and removes the head of this queue, waiting up to the specified wait time if necessary for an element to become available.

*  取出然后删除队列头元素，与上面方法的不同点在于此方法指定了阻塞的时间

## 小结 ##

这里对阻塞队列的API做一个小结

<table> 
 <thead> 
  <tr> 
   <th>存数据操作</th> 
   <th>是否会阻塞</th> 
   <th>是否可以指定阻塞时间</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>add(E e)</td> 
   <td>不会</td> 
   <td>不可以</td> 
  </tr> 
  <tr> 
   <td>offer(E e)</td> 
   <td>不会</td> 
   <td>不可以</td> 
  </tr> 
  <tr> 
   <td>offer(E e, long timeout…)</td> 
   <td>会</td> 
   <td>可以</td> 
  </tr> 
  <tr> 
   <td>put(E e)</td> 
   <td>会</td> 
   <td>不可以</td> 
  </tr> 
 </tbody> 
</table>

例如，我们需要只阻塞一段时间，那就需要使用带时限的offer方法，若想要阻塞直到可以放入元素，那就需要put或者带时限的offer，如果不想阻塞，只想尝试put一下，可以使用offer方法，add方法不推荐使用。

<table> 
 <thead> 
  <tr> 
   <th>取数据操作</th> 
   <th>是否会阻塞</th> 
   <th>是否可以指定阻塞时间</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>take()</td> 
   <td>会</td> 
   <td>不可以</td> 
  </tr> 
  <tr> 
   <td>poll(long timeout…)</td> 
   <td>会</td> 
   <td>可以</td> 
  </tr> 
 </tbody> 
</table>

取元素操作相对比较简单一些，若不想要阻塞，只是尝试获取，可以使用poll(0)

# ArrayBlockingQueue #

这个阻塞队列是一个有界队列，其界限特性在其构造函数中就可以看出

public ArrayBlockingQueue(int capacity) { 
      this(capacity, false);
    }
    
    public ArrayBlockingQueue(int capacity, boolean fair) { 
      if (capacity <= 0)
        throw new IllegalArgumentException();
      // 内部的数据结构，数组
      this.items = new Object[capacity];
      // 阻塞的特性的实现，类似获取特定锁下的wait-notify操作
      lock = new ReentrantLock(fair);
      notEmpty = lock.newCondition();
      notFull =  lock.newCondition();
    }

由此可见，在创建一个ArrayBlockingQueue时必须指定一个容量，队列内部的数据结构则为一个数组来存放

/** The queued items */
    final Object[] items;

很简单，在队列初始化之后会分配一段连续的内存（数组），以数组作为其数据结构。下面来看看开头我们提到的API都是怎么实现的

其中add方法为父类AbstractQueue的模版方法

public boolean add(E e) { 
      if (offer(e))
        return true;
      else
        throw new IllegalStateException("Queue full");
    }

其实就只是offer而已，只不过add会抛出异常，建议使用offer方法，省去处理异常这一步

## offer（无时限） ##

public boolean offer(E e) { 
      checkNotNull(e);
      final ReentrantLock lock = this.lock;
      // 因为要使用condition（wait-notify）特性，所以需要获取锁
      // 重复逻辑下面都会出现，不再赘述
      lock.lock();
      try { 
        // 当前容量等于数组长度，代表队列满了，直接返回false
        if (count == items.length)
          return false;
        else { 
          // 元素入队
          enqueue(e);
          return true;
        }
      } finally { 
        lock.unlock();
      }
    }

很简单，由于offer(E e)方法不具有阻塞特性，所以在队列满的时候直接返回false

## offer（有时限） ##

public boolean offer(E e, long timeout, TimeUnit unit)
      throws InterruptedException { 
    
      checkNotNull(e);
      long nanos = unit.toNanos(timeout);
      final ReentrantLock lock = this.lock;
      // 值得一提，这里可响应中断
      lock.lockInterruptibly();
      try { 
        // 如果队列满了
        while (count == items.length) { 
          if (nanos <= 0)
            return false;
          // 则等待一个限定的时间
          nanos = notFull.awaitNanos(nanos);
        }
        // 入队
        enqueue(e);
        return true;
      } finally { 
        lock.unlock();
      }
    }

没什么好说的，时限等待是通过Condition的实现来做的。这里enqueue是通用入队方法，在后面再详细分析

## put(E e) ##

public void put(E e) throws InterruptedException { 
      checkNotNull(e);
      final ReentrantLock lock = this.lock;
      lock.lockInterruptibly();
      try { 
        // 如果队列满了
        while (count == items.length)
          // 类似wait方法，等待有线程向队列添加元素，就会唤醒
          notFull.await();
        // 入队
        enqueue(e);
      } finally { 
        lock.unlock();
      }
    }

看到这里，可以看出来，其入队的重点就在enqueue方法中

private void enqueue(E x) { 
      // assert lock.getHoldCount() == 1;
      // assert items[putIndex] == null;
      final Object[] items = this.items;
      // putIndex是一个游标
      items[putIndex] = x;
      if (++putIndex == items.length)
        // 当队列满时重置为0，因为是数组，且先进先出的原则
        putIndex = 0;
      count++;
      // 唤醒阻塞在因为线程为空而获取不到元素的线程
      notEmpty.signal();
    }

取元素的操作例如poll、take方法也都是很简单的，读者可以自行查看，关键分析dequeue方法

private E dequeue() { 
      // assert lock.getHoldCount() == 1;
      // assert items[takeIndex] != null;
      final Object[] items = this.items;
      @SuppressWarnings("unchecked")
      // 获取此时游标上对应的元素
      E x = (E) items[takeIndex];
      // 注意需要被remove
      items[takeIndex] = null;
      if (++takeIndex == items.length)
        takeIndex = 0;
      count--;
      if (itrs != null)
        itrs.elementDequeued();
      // 唤醒那些因为队列满了而存不进元素阻塞的线程
      notFull.signal();
      return x;
    }

## 小结 ##

其阻塞特性，大致是如下方式实现的：

*  取元素时若没元素，则在notEmpty这个Condition上等待，取完元素就会在notFull这个Condition上唤醒线程
 *  存元素时若没元素，则在notFull这个Condition上等待，存完元素就会在notEmpty这个Condition上唤醒线程

可以看到，其使用了两个条件变量Condition，去控制阻塞的行为，很好的封装了Condition的使用，使得我们可以在生产-消费模型中直接拿来使用

ArrayBlockingQueue阻塞队列总结如下：

*  有界队列，需要指定大小
 *  数组结构，初始化时需要分配一段连续的内存

# LinkedBlockingQueue #

老样子，先看看其构造函数的实现

public LinkedBlockingQueue() { 
      this(Integer.MAX_VALUE);
    }
    
    public LinkedBlockingQueue(int capacity) { 
      if (capacity <= 0) throw new IllegalArgumentException();
      this.capacity = capacity;
      // 初始化一个Node
      last = head = new Node<E>(null);
    }

可以看到，如果在构造函数中指定一个容量，则此队列就是有界的，如果没有指定容量，可以视为无界队列。

其使用AtomicInteger来存放容量

/** Current number of elements */
    private final AtomicInteger count = new AtomicInteger();

其数据结构为一串单向链表

static class Node<E> { 
      // 元素内容
      E item;
      // 下一个节点
      Node<E> next;
      Node(E x) {  item = x; }
    }

以put为例看看存放元素是如何实现的

public void put(E e) throws InterruptedException { 
      if (e == null) throw new NullPointerException();
      int c = -1;
      // 构造一个Node节点
      Node<E> node = new Node<E>(e);
      final ReentrantLock putLock = this.putLock;
      // 获取容量
      final AtomicInteger count = this.count;
      putLock.lockInterruptibly();
      try { 
        // 满了的话就阻塞
        while (count.get() == capacity) { 
          notFull.await();
        }
        // 入队
        enqueue(node);
        // 增加容量
        c = count.getAndIncrement();
        // 当前容量是否小于界限容量
        if (c + 1 < capacity)
          notFull.signal();
      } finally { 
        putLock.unlock();
      }
      if (c == 0)
        signalNotEmpty();
    }

阻塞特性依旧使用Condition实现，不多赘述，存取元素的关键都在enqueue、dequeue方法

private void enqueue(Node<E> node) { 
      // 在尾节点的next塞节点
      last = last.next = node;
    }

private E dequeue() { 
      // assert takeLock.isHeldByCurrentThread();
      // assert head.item == null;
      Node<E> h = head;
      Node<E> first = h.next;
      h.next = h; // help GC
      head = first;
      E x = first.item;
      first.item = null;
      return x;
    }

很简单，入队不过是在尾部next增加一个节点，出队不过是取出头节点

## 小结 ##

这里就不做过多分析了，都是一些重复的逻辑，其阻塞特性也是使用了两个Condition去实现的。

LinkedBlockingQueue阻塞队列总结如下：

*  可有界可无界，由构造函数的参数决定
 *  链表结构，好处在于初始化时不用向数组那样要先分配一段连续的内存

# SynchronousQueue #

接下来就是最难的传球手队列，其本质不存放元素，所以它的存取方法都比较有特点。先来看看构造函数

public SynchronousQueue() { 
      // 非公平栈
      this(false);
    }
    
    public SynchronousQueue(boolean fair) { 
      // 默认为非公平的实现
      transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
    }

transferer变量是实现存取的关键，其默认使用非公平实现，也就是TransferStack

此队列几个存取方法都一样调用了transferer变量的transfer方法，例如offer方法

public boolean offer(E e) { 
      if (e == null) throw new NullPointerException();
      return transferer.transfer(e, true, 0) != null;
    }

只不过transfer方法的参数不同，这里总结一个表格，然后详细分析transfer方法，期间读者可以对照表格来看看transfer方法的一系列不同行为

<table> 
 <thead> 
  <tr> 
   <th>存取方法</th> 
   <th>transfer方法调用格式</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>put(E e)</td> 
   <td>transferer.transfer(e, false, 0)</td> 
  </tr> 
  <tr> 
   <td>offer(E e, long timeout…)</td> 
   <td>transferer.transfer(e, true, unit.toNanos(timeout)</td> 
  </tr> 
  <tr> 
   <td>offer(E e)</td> 
   <td>transferer.transfer(e, true, 0)</td> 
  </tr> 
  <tr> 
   <td>take()</td> 
   <td>transferer.transfer(null, false, 0)</td> 
  </tr> 
  <tr> 
   <td>poll(long timeout…)</td> 
   <td>transferer.transfer(null, true, unit.toNanos(timeout))</td> 
  </tr> 
  <tr> 
   <td>poll()</td> 
   <td>transferer.transfer(null, true, 0)</td> 
  </tr> 
 </tbody> 
</table>

可以发现一个规律，一定会阻塞的方法在第二个参数都为false，不会阻塞或只阻塞一段时间的第二个参数都为true，第三个参数则为阻塞限定时间（如果有的话），第一个参数为null表示是取元素

在分析之前，先来看一下关键的transferer对象的结构：

/** Dual stack */
    static final class TransferStack<E> extends Transferer<E> { 
      
      // 有三种模式
      /* Modes for SNodes, ORed together in node fields */
      /** Node represents an unfulfilled consumer */
      // 请求数据模式，例如take
      static final int REQUEST    = 0;
      /** Node represents an unfulfilled producer */
      // 插入数据模式，例如put
      static final int DATA       = 1;
      /** Node is fulfilling another unfulfilled DATA or REQUEST */
      static final int FULFILLING = 2;
      
      // 其还有一个SNode对象，作为栈结构的头节点
      /** The head (top) of the stack */
      volatile SNode head;
    }

还有一个关键对象，就是上面代码里的head的那个SNode对象：

static final class SNode { 
      // 可以看出，虽说是栈结构，其内部也很像一个单向链表
      // 这里保存了一个指向了下一个节点的引用
      volatile SNode next;        // next node in stack
      volatile SNode match;       // the node matched to this
      // 保存该节点所属的线程，为了唤醒线程所以需要保存一个
      volatile Thread waiter;     // to control park/unpark
      // 如果为null，表示此节点是取模式，如果有数据，表示此节点是存模式
      Object item;                // data; or null for REQUESTs
      // 模式，在TransferStack对象中有声明
      int mode;
    }

## 有一个线程调用put方法 ##

为了降低复杂度，首先我们来模拟一个流程，假设有一个线程正在对队列调用put方法，准备插入元素。由上面的表格可知，put方法的参数为e, false, 0，分别表示：插入元素、不超时（一直阻塞）、无超时时间

// e, false, 0
    E transfer(E e, boolean timed, long nanos) { 
      SNode s = null; // constructed/reused as needed
      // mode = DATA 表示插入模式
      int mode = (e == null) ? REQUEST : DATA;
    
      for (;;) { 
        // 此时是刚开始，所以 h = head = null
        SNode h = head;
        // 进入此分支
        if (h == null || h.mode == mode) {   // empty or same-mode
          // 超时判断在这里体现，timed就代表是否有超时限制，有超时限制且nanos=0在此就会立即返回了
          // 这里为false，则代表没有超时的限制
          if (timed && nanos <= 0) {       // can't wait
            if (h != null && h.isCancelled())
              casHead(h, h.next);     // pop cancelled node
            else
              return null;
          } 
          // 进入这条分支，对当前put请求构造一个SNode对象
          // 当前SNode表示item=e，next=null，mode=存模式
          // 然后将此SNode设置为head
          else if (casHead(h, s = snode(s, e, h, mode))) { 
            // 阻塞线程
            SNode m = awaitFulfill(s, timed, nanos);
            // 这里埋下伏笔，如果m和s相等，代表节点此时是被取消或中断了
            // 所以如果要取消一个节点的等待，可以唤醒并将awaitFulfill返回值设置为自身
            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
            // 因为是插入模式 DATA，所以此时返回s节点的item
            return (E) ((mode == REQUEST) ? m.item : s.item);
          }
        } 
        // 省略无关代码路径...
      }
    }

此时的流程如下所示：

1.  没有超时限制，则会走下面流程（因为此时仅仅只有put方，没有take方）
2.  构造一个SNode，将当前节点设置为头节点head，模式为存DATA模式
3.  awaitFulfill方法阻塞当前线程
4.  直到有取节点时，会唤醒当前阻塞的线程，然后就可以返回了

其中3、4的步骤还是比较模糊的，其中第四个步骤在分析了take流程就懂了，下面来分析一下第三个步骤，awaitFulfill方法如何阻塞线程

// s为上面构造好的SNode，timed=false，nanos=0
    SNode awaitFulfill(SNode s, boolean timed, long nanos) { 
      // 从这里可以看出，如果timed=true有超时限制，此时就会计算一个超时时间
      final long deadline = timed ? System.nanoTime() + nanos : 0L;
      Thread w = Thread.currentThread();
      // 判断是否需要自旋
      int spins = (shouldSpin(s) ?
                   (timed ? maxTimedSpins : maxUntimedSpins) : 0);
      for (;;) { 
        // 检查中断标志
        if (w.isInterrupted())
          // 如果被中断，需要取消节点
          // 刚刚也说了，取消节点其实就是将当前SNode对象放入自身的match变量中
          s.tryCancel();
        SNode m = s.match;
        // 如果match变量不为null，有两种可能，其中一种就是上面说的中断
        if (m != null)
          // 如果是中断，此时会返回当前SNode
          return m;
        // 有超时限制
        if (timed) { 
          nanos = deadline - System.nanoTime();
          if (nanos <= 0L) { 
            // 超过了超时时间，直接取消
            s.tryCancel();
            continue;
          }
        }
        // 如果需要自旋，则自旋参数-1，继续循环
        if (spins > 0)
          spins = shouldSpin(s) ? (spins-1) : 0;
        // 接下来即将阻塞线程，所以把当前线程放入SNode变量waiter中，方便后面其他线程可以唤醒本线程
        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);
      }
    }

到这里可以看出几个特点：

1.  自旋特性（当前节点为头节点就有可能自旋）
2.  跳出阻塞圈的关键就在SNode的match变量是否为null，这里我们只看到了取消节点时跳出循环的情况，还没有看到别的情况，暂时存疑，在后面揭晓
3.  调用LockSupport的阻塞方法，阻塞当前线程

到这里，我们假设当前线程就已经被阻塞住了，接下来我们继续假设，又有一个线程调用了同样的put方法，那么会发生什么呢？

E transfer(E e, boolean timed, long nanos) { 
      SNode s = null; // constructed/reused as needed
      // DATA
      int mode = (e == null) ? REQUEST : DATA;
    
      for (;;) { 
        // 此时head为刚刚的SNode
        SNode h = head;
        // head的mode显然和此时的put一样，进入该分支
        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;
          }
          // 同样是进入此分支，构造当前SNode成为head节点
          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);
          }
        } 
        // ...
      }
    }

可以看到，第二个线程的put也会导致等待，值得一提的是第二个线程成为了head，就像入栈一样的操作。此时类结构如下图所示![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzM3NzE2_size_16_color_FFFFFF_t_70]  
可以假设，如果后面再有线程进来put，重复入栈操作，成为head

## 另一个线程调用take方法 ##

假设此时有一条线程想要从队列中取元素，于是它调用了transferer.transfer(null, false, 0)

// null, false, 0
    E transfer(E e, boolean timed, long nanos) { 
      SNode s = null; // constructed/reused as needed
      // REQUEST
      int mode = (e == null) ? REQUEST : DATA;
    
      for (;;) { 
        // B Snode
        SNode h = head;
        // 模式不一样且不为null，不进入此分支
        if (h == null || h.mode == mode) {   // empty or same-mode
           // ...
        } 
        // 进入此分支
        else if (!isFulfilling(h.mode)) {  // try to fulfill
          if (h.isCancelled())            // already cancelled
            casHead(h, h.next);         // pop and retry
          // 假设没有被取消，此时会构造一个SNode，next=刚刚的head，模式为FULFILLING
          else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) { 
            for (;;) {  // loop until matched or waiters disappear
              // m = 刚刚的 head = B SNode
              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
              }
              // mn = A SNode
              SNode mn = m.next;
              // unpark B SNode对应的线程，下面会提到
              if (m.tryMatch(s)) { 
                // 将A SNode 设置为head
                casHead(s, mn);     // pop both s and m
                // mode没有被修改过，此时mode = REQUEST，返回 m的item也就是B SNode的元素
                return (E) ((mode == REQUEST) ? m.item : s.item);
              } else                  // lost match
                s.casNext(m, mn);   // help unlink
            }
          }
        } else {                             // help a fulfiller
          // ...
        }
      }
    }

以上流程可以简化如下：

1.  唤醒刚刚的head也就是B SNode，并取其item（要传递的元素）给当前take线程
2.  将A SNode设置为head
3.  如果下面还有线程过来take，以此类推还会唤醒A SNode，然后把head设置为null表示没有元素了

以上流程唯一的疑点就在唤醒线程的tryMatch方法，从注释中可以看出，是B SNode调用了此方法，参数为take线程的那个SNode

boolean tryMatch(SNode s) { 
      // 判断是否match过，没有冲突即为null
      // 然后将当前take线程的SNode置换为B SNode的match变量，这样B线程就可以从之前那个阻塞的循环退出了
      if (match == null &&
          UNSAFE.compareAndSwapObject(this, matchOffset, null, s)) { 
        // 取出B线程
        Thread w = waiter;
        if (w != null) {     // waiters need at most one unpark
          waiter = null;
          // 唤醒B线程
          LockSupport.unpark(w);
        }
        return true;
      }
      return match == s;
    }

很简单，此方法主要就是唤醒了线程，并且将SNode的match设置了一下，这样可以让B线程从awaitFulfill方法中的循环中退出，忘记了的读者可以回忆一下这个方法。此时队列状态如下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzM3NzE2_size_16_color_FFFFFF_t_70 1]  
这个时候聪明的读者可能已经发现了，无论是非阻塞的poll方法还是阻塞一段时间的poll还是刚刚的阻塞方法take，其实在head节点不为null也就是已经有线程在等待put元素的前提下都是一样的，换句话说，transfer(E e, boolean timed, long nanos)方法的三个参数，此时只有第一个参数有用，表示是REQUEST取模式，那么后两个参数有什么用呢？可以假设此时栈无元素，head=null，假设此时有线程来取元素，调用了poll(E e) 不阻塞的方法

// null, true, 0
    E transfer(E e, boolean timed, long nanos) { 
      
      SNode s = null; // constructed/reused as needed
      // REQUEST
      int mode = (e == null) ? REQUEST : DATA;
    
      for (;;) { 
        SNode h = head;
        // head=null，进入此分支
        if (h == null || h.mode == mode) {   // empty or same-mode
          // time=true，有超时限制且超时时间为0，表示不阻塞
          if (timed && nanos <= 0) {       // can't wait
            if (h != null && h.isCancelled())
              casHead(h, h.next);     // pop cancelled node
            else
              // 直接返回null
              return null;
          } 
        } 
        // ...
      }
    }

可以看到，此时是会直接返回的，这个情况放在不阻塞的offer方法也是如此，从开头表格来看，offer方法的调用签名为transferer.transfer(e, true, 0)

// e, true, 0
    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;
        // 进入此分支，因为head=null
        if (h == null || h.mode == mode) {   // empty or same-mode
          // timed=true，有超时限制且超时时间为0
          if (timed && nanos <= 0) {       // can't wait
            if (h != null && h.isCancelled())
              casHead(h, h.next);     // pop cancelled node
            else
              // 直接返回null
              return null;
          } 
          //...
      }
    }

由此我们可以总结SynchronousQueue这个队列的几个入队出队特性：

*  取元素
    
     *  阻塞
        
         *  take：一直阻塞到有线程put元素为止，若多线程take，则依次入栈（不公平）
         *  poll(long timeout…)：一直阻塞直到有线程put元素或者等待了timeout就会唤醒
     *  非阻塞
        
         *  poll()：如果没有线程put，直接返回，此方法需要有线程在阻塞put，才可以获取到元素
 *  存元素
    
     *  阻塞
        
         *  put：一直阻塞到有线程take元素为止，若多线程put，则依次入栈（不公平）
         *  offer(long timeout…)：一直阻塞直到有线程take元素或者等待了timeout就会唤醒
     *  非阻塞
        
         *  offer()：如果没有线程take，直接返回，此方法需要有线程在阻塞take，才可以获取到元素

## 小结 ##

可以看到，此队列很特殊，其本身并不存放元素，其中的数据结构更像是一个单向链表，每一个节点中存放要传递的元素，以生产者的角度来看，没人接收我要放的元素我就会一直等待，这个特性是在阻塞队列中是比较特殊的。

又因为其本身并不存储元素，只是非阻塞的一接一收的特性，使得这种队列在吞吐量方面会优于以上两种队列，所以在高并发低耗时的生产-消费场景下，此队列的吞吐量表现的比较优秀。

当然，以上分析的是非公平的Transferer实现，读者也可以去了解一下公平的实现TransferQueue。

此队列还有一些细节没有分析到，例如transfer还有第三个分支，帮助分支，线程还会帮助别的线程去唤醒和置换head操作，这是并发大师经常玩的套路，这种思想在ConcurrentHashMap中也存在，多线程在大师手里可谓是非常灵活，不愧具有高吞吐量、高伸缩性的特性。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzM3NzE2_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/1c94b2d534654a46a1b87f01e54fe000.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQxNzM3NzE2_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/240526dcb6e647b2bf2f4a58dfb0a467.png