【JUC源码】阻塞队列：SynchronousQueue 源码分析（非公平模式）

╰半夏微凉° 2022-12-11 07:50 200阅读 0赞

> 阻塞队列系列：
> 
>  *  [【JUC源码】阻塞队列：继承关系及接口分析][JUC]
>  *  [【JUC源码】阻塞队列：ArrayBlockingQueue 源码分析][JUC_ArrayBlockingQueue]
>  *  [【JUC源码】阻塞队列：LinkedBlockingQueue 源码分析][JUC_LinkedBlockingQueue]
>  *  [【JUC源码】阻塞队列：DelayQueue 源码分析][JUC_DelayQueue]
>  *  [【JUC源码】阻塞队列：SynchronousQueue 源码分析（非公平模式）][JUC_SynchronousQueue]
>  *  [【JUC源码】阻塞队列：SynchronousQueue 源码分析（公平模式）][JUC_SynchronousQueue 1]
>  *  [【JUC源码】阻塞队列：总结 --特性对比&应用场景举例][JUC_ --]
>  *  [【JUC源码】阻塞队列：各种阻塞队列面试题合集][JUC 1]

SynchronousQueue 同步队列，队列不存储数据，所以没有大小，也无法迭代。没有大小如何理解呢？

*  每次进行put值进去时, 必须等待相应的 consumer 拿走数据后才可以再次 put 数据。
 *  queue 对应 peek, contains, clear, isEmpty … 等方法其实是无效的。

SynchronousQueue 继承关系，核心成员变量及主要构造函数：

public class SynchronousQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
        
    	
    	// Transferer 定义了transfer方法，put，take都是用的同一个transfer方法
    	abstract static class Transferer<E>{
        
    	    // e为空的，会直接返回特殊值，不为空会传递给消费者
            abstract E transfer(E e, boolean timed, long nanos);
    	}
    	
    	// 堆栈实现，后入先出（非公平）
    	static final class TransferStack<E> extends Transferer<E>{
        ...}
    	
    	// 队列实现，先入先出（公平）
    	static final class TransferQueue<E> extends Transferer<E>{
        ...}
    	
    	// tranfer变量
    	private transient volatile Transferer<E> transferer;
    	
    	//----------------------------------构造函数---------------------------------
    	// 默认非公平
    	public SynchronousQueue() {
        
            this(false);
        }
        
        // 公平用TransferQueue，非公平就用TransferStack
        public SynchronousQueue(boolean fair) {
        
            transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
        }
        
    }

这里需要着重强调的是，SynchronousQueue 自己实现了对线程的控制，并**没有使用锁**(synchronized 与 reentrantlock) ，因为锁底层的队列无法实现匹配。所以 SynchronousQueue 必须自己保证线程安全，并实现线程的调度：

*  通过**CAS与自旋**实现线程安全
 *  **直接对线程进行阻塞（park）与唤醒**
 *  两种策略，堆栈实现的非公平配对（FILO），与队列实现的公平配对（FIFO）

## 1.TransferStack（非公平=>FILO） ##

static final class TransferStack<E> extends Transferer<E>{
        
       
            // 栈中元素，这是一个链式栈
            static final class SNode{
        ...}
            // 栈头指针
            volatile SNode head;
    		
    		// SNode的三种状态：
        	// 1.REQUEST：执行的是take方法，相当于消费者
            static final int REQUEST    = 0;
            // 2.DATA：执行的是put方法，相当于生产者
            static final int DATA       = 1;
            // 3.FULFILLING：栈头正在阻塞等待其他线程进行 put 或 take
            static final int FULFILLING = 2;
        
        	//...
    }

### SNode ###

虽然 SynchronousQueue 的特性说的是里面是没有元素，但这句话实际的意义是 SynchronousQueue 是一个不能 peek，contains 等操作的节点，但是里面是有一条链表来**保存竞争的线程和数据**。

也就是说，整个 SynchronousQueue 的运行机制也还是通过维护一个链表来实现的。当有并发时，通过判断链表插入一端节点的类型（mode），从而确定是否进行交换。

static final class SNode {
        
                
                // 当前线程
                // 注：不是在创建SNode时设置，而是在awaitFulfill方法中没匹配到需要休眠时才会设置
                volatile Thread waiter;
                // 当前线程数据
                // 注：只有put的线程item才会有值，take的线程item=null。若take的线程要取出数据只能通过match指针为中介
                Object item;
        	    // 节点类型：REQUEST(0)-消费者（take） ，DATA(1)-生产者（put）， FULFILLING(2)-交换中
                int mode;
    			
    			volatile SNode next;
                // 很重要的节点。表示和本节点配对（match）的节点，有两个作用：
                // 1.判断阻塞栈元素能被唤醒的时机
                //  比如线程A在take时由于队列为空被阻塞了，然后线程B进行了put操作，那么就会将A的match设置为B，表示可以将A唤醒了
                // 2.作为take线程获取数据交换的中介
                //  比如当线程A唤醒后要返回数据，那么就可以通过match找到B，从而拿到put的数据。这个逻辑可以在下面的transfer方法可以看到
                volatile SNode match;
    ------------------------------------------------------------------------------------------------------------------			
    			// 构造函数，传入item
    			// 注：一般不直接构造，而是调用封装好的snode方法：SNode snode(SNode s, Object e, SNode next, int mode) 
                SNode(Object item) {
        
                    this.item = item;
                }
    ------------------------------------------------------------------------------------------------------------------			
        	    // 通过CAS把val节点连接到cmp后面
                boolean casNext(SNode cmp, SNode val) {
        
                    return cmp == next &&
                           UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
                }
    			
        		// tryMatch 非常重要的方法，两个作用：
                // 1 尝试将参数节点s，赋给当前线程的配对节点match
                // 2.唤醒被阻塞的栈头的线程，醒后就能从 match 中得到本次操作 s
                // 其中 s.item 记录着本次的操作节点，也就是记录本次操作的数据
                boolean tryMatch(SNode s) {
        
                    if (match == null &&
                        // CAS改变match来进行匹配，成功的条件是当前节点的match属性为null
                        UNSAFE.compareAndSwapObject(this, matchOffset, null, s)) {
        
                        Thread w = waiter;
                        if (w != null) {
            
                            // 将waiter置为null
                            waiter = null;
                            // 唤醒当前node的线程
                            LockSupport.unpark(w);
                        }
                        return true;
                    }
                    // 返回是否配对成功
                    return match == s; 
                }
    
               // 尝试取消，就是把match换为自己。
    		   // 自己匹配自己的前提是match为null，也就是说，如果已经匹配了，那么这个方法不能取消
        	   // 一般在设置超时且过期后，会将其设为cancel
                void tryCancel() {
        
                    UNSAFE.compareAndSwapObject(this, matchOffset, null, this);
                }
        
    		   // 判断超时失效
        	   // 和上一个方法对比，则可以知道，判断match是不是自己
                boolean isCancelled() {
        
                    return match == this;
                }
    
               // unsafe相关代码...
    }

### transfer()：进栈&出栈 ###

transfer 将 take 和 put 两个方法都揉在了一起，所以第一个问题是如何区分 put 和 take？是通过参数 e 是否为null来区分，e不为 null 是 put，为null是 take。

第二个问题是如何进行线程管理的，或者说如何使 put 和 take 的线程配对的？具体分为以下三种情况：

*  情况 1：队列中还没有数据 或 当前节点与栈顶节点同类型（同put或同take）
    
     *  情况 1.1：要加入的e设置了超时时间，并且 e 进栈或者出栈要超时了
        
         *  情况 1.1.1：栈头不为null 且 栈头已经超时失效，将栈顶置为第二个节点
         *  情况 1.1.2：栈头是空的，返回null
     *  情况 1.2：没有设置新元素e的超时时间，或者设置了但未超时，
        
        1.  用e构造新节点s，使s.next=head，然后将s设为新的栈头
        2.  阻塞等待与s匹配的节点m
        3.  若没等到（s过期了），就调用clean删除s
        4.  若等到了，就让s与m出栈，设置新的头结点，并返回
 *  情况 2： 当前栈包含于给定节点模式互补的节点（比如栈顶是put时阻塞，而当前节点是take操作）
    
     *  情况 2.1：栈头已经被取消，将下一个节点置为栈头
     *  情况 2.2：可以将当前节点s打上"正在匹配"的标记，并设置为head
        
        1.  取s的下一个节点m，s与m不断tryMatch
        2.  匹配成功，删除s与m，返回item
        3.  匹配失败，交换m与m.next
        4.  若栈完了都没匹配到则退出自旋，进入3
 *  情况 3：当动作2匹配失败（可能同时线程3抢先完成了配对），帮助这个节点完成匹配和移除(出栈)的操作。然后继续执行(主循环)。这部分代码基本和动作2的代码一样，只是不会返回节点的数据。

@SuppressWarnings("unchecked")
    E transfer(E e, boolean timed, long nanos) {
        
        SNode s = null; 
        // 判断是put还是take：e 为空是take方法(REQUEST)，不为空是put方法(DATA)
        int mode = (e == null) ? REQUEST : DATA;
        // 自旋，保证一定能成功
        for (;;) {
        
            // 拿出头节点，有几种情况
            SNode h = head;
    -----------------------------------------------------------------------------------------------------------------        
            // 情况 1：队列中还没有数据 || 当前节点与栈顶节点同类型（同put或同take）
            if (h == null || h.mode == mode) {
        
                // 情况 1.1：要加入的e设置了超时时间，并且 e 进栈或者出栈要超时了
                if (timed && nanos <= 0) {
             
                    // 情况 1.1.1：栈头不为null && 栈头已经超时失效
                    if (h != null && h.isCancelled())
                        casHead(h, h.next); // 丢弃栈头，把栈头后一个元素作为栈头
                    // 情况 1.1.2：栈头是空的
                    else
                        return null; // 直接返回 null
                // 情况 1.2：没有设置新元素e的超时时间，或者设置了但未超时
                } else if (casHead(h, s = snode(s, e, h, mode))) {
         // 用e构造新节点s，使s.next=head，然后将s设为新的栈头
                    // 阻塞等待，目的是等到与s匹配的SNode
                    SNode m = awaitFulfill(s, timed, nanos);
                    // 返回m==s代表当前节s点已经超时了
                    if (m == s) {
           
                        clean(s); // 栈中无法直接删除s，所以调用clean
                        return null;
                    }
                    // 只有真正匹配到值才能走到这一步
                    // 栈不为空 && 栈二是s（因为等到的m此时是新栈顶）
                    if ((h = head) != null && h.next == s)
                        // 将s.next设置为head，表示将s和他的配对m出栈
                        casHead(h, s.next);  
                    // 返回item
                    // 注：这里返回的是put的节点的数据，若是take节点那么需要通过中介m来获取到数据
                    return (E) ((mode == REQUEST) ? m.item : s.item);
                }
    -----------------------------------------------------------------------------------------------------------------                    
            // 情况 2： 当前栈包含于给定节点模式互补的节点（比如栈顶是put时阻塞，而当前节点是take操作）
            } else if (!isFulfilling(h.mode)) {
         
                // 情况 2.1：栈头已经被取消
                if (h.isCancelled())            
                    casHead(h, h.next); // 把下一个元素作为栈头
                // 情况 2.2：可以将当前节点s打上"正在匹配"的标记，并设置为head
                else if (casHead(h, s=snode(s, e, h, FULFILLING|mode))) {
        
                    // 自旋，直到配对
                    for (;;) {
         
                        // m和s是正在匹配两个节点
                        // 注：此时m不一定是之前的栈顶，因为这段时间可能又有节点进入或者之前的栈顶已被先一步匹配走了
                        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
                        }
                        // 获取m的next节点，因为如果s和m匹配成功，mn就得补上head的位置了
                        // 注：虽然后面m可能会与mn交换，但在每轮循环中mn一定是栈中第三个节点
                        SNode mn = m.next;
                        // 调用 tryMatch 让 m和s 配对
                        // 注：这里调用tryMatch给s配对时，会唤醒阻塞在awaitFulfill方法的线程m，若配对失败m还会回到阻塞状态
                        if (m.tryMatch(s)) {
        
                            // 配对成功，弹出s与m，将head置为mn
                            casHead(s, mn);     
                            // 返回put的数据，若是take节点那么需要通过中介m来获取到数据
                            return (E) ((mode == REQUEST) ? m.item : s.item);
                        } else  
                            // 若配对失败，将m与m.next交换，开始下一轮匹配
                            s.casNext(m, mn);   
                    }
                }
    -----------------------------------------------------------------------------------------------------------------                    
           	// 情况3：上面匹配失败，可能是同时又线程3提前完成了配对  
            } else {
                                     
                SNode m = h.next;  
                // 栈里面没有任何等待者了,其他节点把m匹配走了
                if (m == null)                 
                    casHead(h, null);           // pop fulfilling node
                else {
        
                    // 如果m和h匹配成功，则mn就成为新head了。
                    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()：等待匹配节点 ###

awaitFulfill 作用是阻塞等待匹配的节点，但**不是一上来就阻塞住，而是在自旋一定次数后**，仍然没有其它线程来满足自己的要求时，才会真正的阻塞住，等待其他线程transfer后tryMatch

1.  计算死亡时间 deadline（时间戳）与自旋次数 spains
2.  自旋，每次循环都判断s是否获得到match
3.  达到自旋次数，park当前线程（定时）
4.  线程被唤醒时机：
    
     *  tryMatch唤醒：其余线程在transfer中遍历栈时调用 tryMatch唤醒，若匹配成功 return 相应 m，否则重新回到阻塞
     *  超时唤醒：return s（自己）

Node awaitFulfill(SNode s, boolean timed, long nanos) {
        
        // deadline 死亡时间，如果设置了超时时间的话，死亡时间等于当前时间 + 超时时间，否则就是 0
        final long deadline = timed ? System.nanoTime() + nanos : 0L;
        // 当前线程
        Thread w = Thread.currentThread();
        // 自旋的次数，如果设置了超时时间，会自旋 32 次，否则自旋 512 次
        int spins = (shouldSpin(s) ? (timed ? maxTimedSpins : maxUntimedSpins) : 0);
                     
        for (;;) {
        
            // 当前线程有无被打断，如果过了超时时间，当前线程就会被打断
            if (w.isInterrupted())
                s.tryCancel();
    		
            // 尝试获取当前节点的match
            SNode m = s.match;
            // 该函数的唯一出口，一定要匹配到值
            // 返回的 m==s 表示超时取消，返回的 m!=s 表确实匹配到了
            if (m != null)
                return m;
            if (timed) {
        
                nanos = deadline - System.nanoTime();
                // 超时了，取消当前线程的等待操作
                if (nanos <= 0L) {
        
                    // 调用cancel，使m=s
                    s.tryCancel();
                    continue;
                }
            }
            
            // 如果没到自旋次数，那么自旋次数减少 1
            if (spins > 0)
                spins = shouldSpin(s) ? (spins-1) : 0; // 
            // 如果s没设置waiter，那么把当前线程设置成 waiter
            else if (s.waiter == null)
                s.waiter = w; 
            // 如果没有设置超时，那么直接park当前线程  
            else if (!timed)
                LockSupport.park(this); // 当被unpark唤醒时也是在此处,继续循环
            // 如果设置了超时，调用带有nanos超时时间的park
            else if (nanos > spinForTimeoutThreshold)
                LockSupport.parkNanos(this, nanos);
        }
    }

### clean()：清除过期节点 ###

清除处于栈中的过期节点s，大致过程如下：

1.  找到s的下一个没有cancel的节点past
2.  判断head是否cancel
3.  从head遍历连接past

void clean(SNode s) {
        
                s.item = null;   // forget item 把item和waiter都置空
                s.waiter = null; // forget thread
    			
                // 获得下一个SNode
                SNode past = s.next;
        	   // 如果past被cancell了，那么就再past一个
                if (past != null && past.isCancelled())
                    past = past.next;
    
                // 从头节点开始清除
                SNode p;
        	    // 把头节点链接到下一个节点，节点不能为cancelled
                while ((p = head) != null && p != past && p.isCancelled())
                    casHead(p, p.next);
    
                // Unsplice embedded nodes
                while (p != null && p != past) {
        
                    // 在去除链接头节点以后的节点，同样也不能为null。
                    SNode n = p.next;
                    if (n != null && n.isCancelled())
                        p.casNext(n, n.next);
                    else
                        p = n;
                }
    }

> PS：由于篇幅限制，关于 TransferQueue 的 FIFO 公平配对模式，请参考[下一篇文章][JUC_SynchronousQueue 1]…

## 2.方法解析 ##

SynchronousQueue 的方法都很简单，因为已经封装好了两种 Transfer 的实现，TransferStack 和 TransferQueue，所以后面的方法直接调用就行。

### 2.1 放入：put ###

将新元素放进队列，直到有另外一个线程从队列中取走，成功结束，失败打断线程

public void put(E e) throws InterruptedException {
        
        // e为空，抛异常
        if (e == null) throw new NullPointerException();
        // 调用transfer方法，传入e
        // 一直等待
        if (transferer.transfer(e, false, 0) == null) {
        
            Thread.interrupted();
            throw new InterruptedException();
        }
    }

下面，我们再看看 offer 方法，offer 不会阻塞，简单点说就是 offer 的时候如果有消费者在等待那么就会立马满足返回true，如果没有就会返回false，不会等待消费者到来

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

可以看到与 put 的区别就是 timed 参数为true。

不会阻塞的原理是（以TransferStack为例）: transfer方法分为两种情况，其中 \[当前栈包含于给定节点模式互补的节点\] 的情况下就不会出现阻塞，若匹配到就会直接返回put的值；而 \[当前节点与栈顶节点类型相同\] 的清况会调用 awaitFulfill 方法阻塞等待互补节点。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]  
从上图可以看到若 timed为true，就只会进入情况 1.1 而不会进入情况1.2中调用 awaitFulfill 方法，所以这种情况下offer也不会阻塞。

### 2.2 取出：take ###

从队列头拿数据并删除数据，成功返回，失败打断线程

public E take() throws InterruptedException {
        
        // 调用transfer方法，传入null
        E e = transferer.transfer(null, false, 0);
        if (e != null)
            return e;
        Thread.interrupted();
        throw new InterruptedException();
    }

下面，我们看看 poll 方法。原理同上面 offer

public E poll() {
        
            return transferer.transfer(null, true, 0);
        }

### 2.3 容量相关方法 ###

与容量相关方法都是默认实现，即写死的。

**peek()**

public E peek() {
        
            return null;
    }

**remove()**

public boolean remove(Object o) {
        
            return false;
    }

**contains()**

public boolean contains(Object o) {
        
            return false;
    }

[JUC]: https://blog.csdn.net/weixin_43935927/article/details/113952868
[JUC_ArrayBlockingQueue]: https://blog.csdn.net/weixin_43935927/article/details/108833582
[JUC_LinkedBlockingQueue]: https://blog.csdn.net/weixin_43935927/article/details/108853656
[JUC_DelayQueue]: https://blog.csdn.net/weixin_43935927/article/details/108858129
[JUC_SynchronousQueue]: https://blog.csdn.net/weixin_43935927/article/details/108866902
[JUC_SynchronousQueue 1]: https://blog.csdn.net/weixin_43935927/article/details/113964294
[JUC_ --]: https://blog.csdn.net/weixin_43935927/article/details/108888294
[JUC 1]: https://blog.csdn.net/weixin_43935927/article/details/108857735
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/0362348e9f7f49919fedd8e285a20d10.png