SynchronousQueue的transfer方法分析（JDK1.8）

我就是我 2023-09-27 10:36 20阅读 0赞

#### 目录 ####

*   **TransferQueue**
 *   *   **TransferQueue原理**
     *   **TransferQueue关键代码分析** 
     *   *   **TransferQueue队列中被阻塞的线程如何被唤醒** 
         *   **TransferQueue线程入队被阻塞** 
     *  TransferQueue的使用
 *   **TransferStack**

在SynchronousQueue中put，offer，poll，take方法都是调用的同一个方法transfer。

//添加元素
        public void put(E e) throws InterruptedException {
        
            if (e == null) throw new NullPointerException();
            if (transferer.transfer(e, false, 0) == null) {
        
                Thread.interrupted();
                throw new InterruptedException();
            }
        }
    
    
        public boolean offer(E e, long timeout, TimeUnit unit)
            throws InterruptedException {
        
            if (e == null) throw new NullPointerException();
            if (transferer.transfer(e, true, unit.toNanos(timeout)) != null)
                return true;
            if (!Thread.interrupted())
                return false;
            throw new InterruptedException();
        }
    
    
        public boolean offer(E e) {
        
            if (e == null) throw new NullPointerException();
            return transferer.transfer(e, true, 0) != null;
        }
    //获取元素
        public E take() throws InterruptedException {
        
            E e = transferer.transfer(null, false, 0);
            if (e != null)
                return e;
            Thread.interrupted();
            throw new InterruptedException();
        }
    
        public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        
            E e = transferer.transfer(null, true, unit.toNanos(timeout));
            if (e != null || !Thread.interrupted())
                return e;
            throw new InterruptedException();
        }
    
    
        public E poll() {
        
            return transferer.transfer(null, true, 0);
        }

存入，获取元素都是调用同一个方法，那么它是如何分辨出方法的调用者到底是存入元素还是要取出元素呢？仔细观察下不难看出在获取，添加元素的时候传入transfer方法的参数是不同的；transfer方法有3个参数，这三个参数分别表示：

*  e：要存放的元素。根据e是否为空可以判断是要添加元素还是要获取元素；
 *  timed（boolean）：该操作是否是超时等待；比如poll(long timeout, TimeUnit unit)就会有超时等待的参数，而take，put没有这个参数，就需要用一个参数来分辨。
 *  超时时间nanos（纳秒）；超时等待的时间；

在SynchronousQueue内部transfer方法有2种实现：TransferQueue，TransferStack。

![在这里插入图片描述][e89822d6ed5f4fab8cbb12721808cbfb.png_pic_center]

##  **TransferQueue** ##

在创建SynchronousQueue对象时可以选择transfer方法的实现类：

![在这里插入图片描述][3de779a32c8848d0b1903e7243680a7e.png_pic_center]

TransferQueue 是一个链表结构，在TransferQueue 内部有一个内部类：QNode，TransferQueue 是由QNode节点构建的链表结构。

![在这里插入图片描述][c93d0a38e07b402a93acc5343507e32c.png_pic_center]

###  **TransferQueue原理** ###

在TransferQueue 创建时会初始化一个QNode节点，head，tail都会指向这个空节点；在TransferQueue中会以根据传入的参数：e是否为null来将节点分成2类 。从TransferQueue队列中获取元素的线程是同一类节点，比如：调用take，poll的线程就是同一类节点；从TransferQueue队列中添加元素的线程是一类节点put，offer是同一类节点；

TransferQueue 队列特殊的地方就在于这个队列中只会存在一种节点：要么是获取元素的线程节点，要么就添加元素的线程节点；

![在这里插入图片描述][355ed27baddc41b99f9895f6d58544ef.png_pic_center]

在初始化TransferQueue对象时，会初始化生成一个节点队列的头，尾：head，tail都会指向这个init节点；

上面说到，在TransferQueue队列中只会存在一种节点也就是说：在队列中的线程要么都是获取元素的，要么都是添加元素的。如果有与队列中不同类型的线程调用transfer方法，就会将阻塞在队列中的节点唤醒。举个例子：假设当前队列中都是put线程，此时有一个take线程；那么这个take线程就会唤醒队列中的一个put线程；  
![在这里插入图片描述][9e4a7c3a337f4544aff77aafeb9a9a95.png_pic_center]

在唤醒线程时，同时会修改该线程所在节点的item值；在后面分析源码的时候会看到，如果只是唤醒线程是没有用的，还需要将item的值修改才能真正唤醒该线程；

通过上面的分析可以知道，TransferQueue是如何实现阻塞以及如何实现唤醒的。当队列为空或者队列中只有一种线程就会将线程阻塞，当出现与队列中不同类型的线程才会将队列中的线程唤醒。这个算法其实不难，看起来比较难的原因主要是在多线程环境下，无锁的实现了线程安全的队列。既然没有锁，那么在多线程环境下队列中的元素随时增减。因此在每一步操作之前都要判断变量值是否被修改了。

###  **TransferQueue关键代码分析**  ###

transfer方法的代码有点长，很多代码都比较简单因此就不全分析了。了解了TransferQueue队列的线程入队出队的原理，看源码也就简单了。

transfer方法中的代码结构，伪代码：

E transfer(E e, boolean timed, long nanos) {
        
                QNode s = null; // constructed/reused as needed
                boolean isData = (e != null);
                for(;;){
        
                	if(队列为空 || 新线程的类型与队列中线程类型一致){
        
                		将线程包装成QNode节点阻塞在队列中
    				}else{
        
    					唤醒队列的节点
    				}
    			}

下面分别分析TransferQueue的出队入队；

####  **TransferQueue队列中被阻塞的线程如何被唤醒**  ####

E transfer(E e, boolean timed, long nanos) {
        
    
                QNode s = null; // constructed/reused as needed
                boolean isData = (e != null);
    
                for (;;) {
        
                    QNode t = tail;
                    QNode h = head;
                    if (t == null || h == null)         // saw uninitialized value
                        continue;                       // spin
    
                    if (h == t || t.isData == isData) {
         
    					h == t ：是空队列，就将线程添加到队列的末尾，然后更新tail；将线程阻塞等待被唤醒；
    					t.isData == isData：新线程与队列中的线程是同一类节点，也将新线程添加到队列末尾，更新tail，将线程阻塞等待被唤醒；
                    } else {
          
                    
                     进入到else分支：说明队列不是空队列，并且 队列中线程与当前线程不是同一类线程；这个时候当前线程会唤醒队列中的一个线程；
                                      
                        QNode m = h.next;               
                        if (t != tail || m == null || h != head)
                            continue;                   
    
                        Object x = m.item;
                        if (isData == (x != null) ||    //后面会单独分析这种情况；
                            x == m ||                    节点被取消，item的值会指向当前节点；
                            !m.casItem(x, e)) {
         这一步操作很关键，将被唤醒的线程的item值修改为当前线程的item
                            advanceHead(h, m);          // dequeue and retry
                            continue;
                        }
    
                        advanceHead(h, m);           修改head的指针，将head指向被唤醒的线程所在节点；           
                        LockSupport.unpark(m.waiter);   唤醒线程；
                   // 无论是put线程唤醒take线程，还是take唤醒put；他们都是成对的；
                   // 由于put线程中item不为null，因此返回值从put线程中获取；被唤醒的线程，返回值必定不为null；
                        return (x != null) ? (E)x : e; 
                    }

进入到else分支说明队列不是空队列，并且 队列中线程与当前线程不是同一类线程；这个时候当前线程会唤醒队列中的一个线程；为什么要获取head的next节点呢？在TransferQueue队列中head节点只是一个占位用的，head节点中并没有被阻塞的线程。队列中被阻塞的节点是从head.next开始的 ，因此在唤醒节点的时候也是从next开始；

**为什么会出现判断：isData == (x != null) ？**

进入到else分支就2种情况：

*  队列中的线程是take线程，当前线程是put线程；
 *  队列中的线程是put线程，当前线程是take线程；

isData是当前线程的类型（获取元素，或者是 存入元素），而x != null是判断队列中线程的类型；就上面的2种情况来看当前线程与队列中的线程类型必然不同，这样才能进入到这个分支中唤醒队列中的线程；

**结果显而易见的应该是false，什么情况下会出现结果为true的情况呢？**

考虑以下情况：有一个take线程被阻塞在队列中，同时有2个put线程唤醒队列中的节点；

![在这里插入图片描述][61a7742026ac401295484a49aaba0c18.png_pic_center]

当thread-B获取到item的值时，item的值已经被修改。此时item != null ===⇒ x != null ;此时 isData == (x != null) 的结果为true；

####  **TransferQueue线程入队被阻塞**  ####

E transfer(E e, boolean timed, long nanos) {
        
    
                QNode s = null; // constructed/reused as needed
                boolean isData = (e != null);
    
    
                for (;;) {
        
                    QNode t = tail;
                    QNode h = head;
                    队列的head，tail会因为线程的出队，入队而更改；在更新值期间，任何线程都不能使用队列；必须等值更新完才能put，take；
                    if (t == null || h == null)         // saw uninitialized value
                        continue;                       // spin
    				h==t:空队列，t.isData == isData新节点的类型与队列中的类型相同；这2种情况下，新节点都会加入到队列中；
                    if (h == t || t.isData == isData) {
         // empty or same-mode
                        QNode tn = t.next;
                        if (t != tail)  队列的尾节点tail已经被更新
                            continue;
                        if (tn != null) {
         有新节点加入到队列中
                            advanceTail(t, tn); 更新tail
                            continue;
                        }
                        if (timed && nanos <= 0)  立即返回，不阻塞在队列中；
                            return null;
                        if (s == null)
                            s = new QNode(e, isData);将线程包装成QNode节点
                        if (!t.casNext(null, s))        将新节点加入到队列尾部
                            continue;
    
                        advanceTail(t, s);            添加成功之后，更新tail，将tail指向刚加入队列的节点
                         //等待被唤醒，有中断标记，或者阻塞线程到设置的阻塞时间；都会从awaitFulfill中被唤醒
                        Object x = awaitFulfill(s, e, timed, nanos);
                        if (x == s) {
        // 中断标记，带阻塞时间的线程等待了规定时间恢复运行          
                            clean(t, s);节点从队列中被清除
                            return null;
                        }
    				
                        if (!s.isOffList()) {
                   该节点脱离队列
                            advanceHead(t, s);          // unlink if head
                            if (x != null)              // and forget fields
                                s.item = s;
                            s.waiter = null;
                        }
                        return (x != null) ? (E)x : e;
                        }else{
        
                       		 唤醒队列中的线程；
    						}
    
    }

阻塞线程的具体操作是在awaitFulfill方法中

Object awaitFulfill(QNode s, E e, boolean timed, long nanos) {
        
                /* Same idea as TransferStack.awaitFulfill */
                final long deadline = timed ? System.nanoTime() + nanos : 0L;//计算需要等待的阻塞时间
                Thread w = Thread.currentThread();//当前线程
                int spins = ((head.next == s) ?
                             (timed ? maxTimedSpins : maxUntimedSpins) : 0);//计算自旋次数
                for (;;) {
        
                    if (w.isInterrupted())//判断线程是否设置了中断标记，如果有中断标记就不阻塞线程；
                        s.tryCancel(e);//修改item的值
                    Object x = s.item;
                    if (x != e)//item的值与刚加入队列时的值不相等；返回
                        return x;
                    if (timed) {
        
                        nanos = deadline - System.nanoTime();
                        if (nanos <= 0L) {
        //已经到了阻塞时间
                            s.tryCancel(e);//修改item的值
                            continue;
                        }
                    }
                    if (spins > 0)//自旋；
                        --spins;
                    else if (s.waiter == null)
                        s.waiter = w;//设置等待线程
                    else if (!timed)
                        LockSupport.park(this);//阻塞线程
                    else if (nanos > spinForTimeoutThreshold)//设置的超时等待时间小于1000纳秒不需要阻塞线程，让线程自旋消耗时间；
                        LockSupport.parkNanos(this, nanos);
                }
            }

需要特别说明一下变量`spins`，所有进入阻塞队列的线程都不着急立即阻塞，而是会先自旋一段时间。自旋的次数就是spins次；为什么会这么设置呢？如果服务器是多核的，那么可以同时运行多个线程。如果并发越高，短时间内就越有可能有新的线程来唤醒队列中的线程；这个时候如果阻塞线程再唤醒线程的代价就比让线程自旋的大。

awaitFulfill方法从for循环中结束的条件就是修改item的值；如果不修改item的值，那么即使线程被唤醒了还是会一直在这个for循环中。有3种情况可以修改item的值：

*  被唤醒；put唤醒take，item：null -> e ; take唤醒put,item: e -> null;
 *  设置了中断异常标记，队列中的线程可能是take可能是put；item：e/null -> this(QNode) ;
 *   过了超时时间恢复运行的线程，item：e/null -> this(QNode);

### TransferQueue的使用 ###

*  put，take的使用（单独使用其中一个都会被阻塞）：

public void test() throws InterruptedException {
        
    
            //SynchronousQueue内部创建一个TransferQueue队列
            SynchronousQueue queue = new SynchronousQueue(true);
    
            for (int i = 0; i < 5; i++) {
        
                int finalI = i;
                new Thread(()->{
        
                    try {
        
                        queue.put("i:"+ finalI);
                        System.out.println(Thread.currentThread().getName()+" : put线程,结束阻塞。。。 ");
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                },"thread-"+i).start();
            }
    
            Thread.sleep(1000);
            System.out.println("=========================put线程被阻塞=======================");
    
            for (int i = 0; i < 5; i++) {
        
                new Thread(()->{
        
                    try {
        
                        queue.take();
                        System.out.println(Thread.currentThread().getName()+":take线程,唤醒了一个put线程。。");
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                },"thread-"+i+i).start();
                Thread.sleep(1000);
            }
        }
    ===========================================================结果===========================================================
    
    =========================put线程被阻塞=======================
    thread-00:take线程,唤醒了一个put线程。。
    thread-0 : put线程,结束阻塞。。。 
    thread-11:take线程,唤醒了一个put线程。。
    thread-1 : put线程,结束阻塞。。。 
    thread-22:take线程,唤醒了一个put线程。。
    thread-2 : put线程,结束阻塞。。。 
    thread-33:take线程,唤醒了一个put线程。。
    thread-3 : put线程,结束阻塞。。。 
    thread-44:take线程,唤醒了一个put线程。。
    thread-4 : put线程,结束阻塞。。。

*  offer，poll的使用（直接使用offer向队列中添加数据会失败）

@Test
        public void test2(){
        
            //SynchronousQueue内部创建一个TransferQueue队列
            SynchronousQueue queue = new SynchronousQueue(true);
    
            //添加值
            boolean putValue = queue.offer("eerrr");
            //获取值
            Object poll = queue.poll();
    
            System.out.println(poll==null);//true,说明没有从队列中获取到值；队列中没有值  =》offer不能将值存到队列中、。
    
    
        }
    ===========================================================结果===========================================================
    true

**为什么offer线程的值别的线程获取不到呢？**

经过上面的分析，这个问题其实很简单。offer(e),在调用transfer方法时的传值：transfer(e,true,0);假设是个空队列或者队列中是添加元素的线程，进入到if分支中，有一个判断条件`if(timed && nanos <=0)return null`，到这里恰好满足判断条件，因此会直接返回null；不将线程阻塞到队列中，后面的poll线程调用transfer方法时该线程已经返回了没有阻塞在队列中，因此获取不到该线程的值。

使用offer(e)的正确方式：如果队列中已经有获取元素的线程被阻塞，此时使用offer被阻塞的线程就可以获取到offer线程的值；

offer还有一个设置等待时间的方法`offer(E e, long timeout, TimeUnit unit)`，在等待时间之内，线程都会被阻塞在队列中。因此使用这个方法，在阻塞时间内，如果有线程获取值，那么就会得到结果。超时之后线程出队，别的线程就获取不到值了。【poll(long timeout, TimeUnit unit) 方法也一样，阻塞时间内没有存入值的线程进入队列就不会获取值；阻塞时间超时之后poll线程就会出队，在阻塞期间没有线程入队自然也就获取不到值了。】

@Test
        public void test3()  throws InterruptedException{
        
            SynchronousQueue queue = new SynchronousQueue(true);
    
    
            new Thread(()->{
        
                try {
        //等待 1s  ；
                    queue.offer("weew",1, TimeUnit.SECONDS);
    //                System.out.println("被唤醒。。");
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }).start();
            
            // Thread.sleep(500);
     		//Object poll = queue.poll();
     		//System.out.println(poll);//weew,在阻塞时间内，因此可以获取值
    		//阻塞1.5s获取不到值
            Thread.sleep(1500);
            Object poll = queue.poll();
            System.out.println(poll);//null
    
        }

SynchronousQueue算是大坑，如果不了解SynchronousQueue就使用的话估计会被坑的很惨。juc下的其他阻塞队列都还行，就算不了解原理的情况使用也不会出大问题。SynchronousQueue没有使用锁，因此效率会比ArrayBlockingQueue，LinkedBlockingDeque，LinkedBlockingQueue这些有锁的队列要高一些。

##  **TransferStack** ##

TransferStack内部有一个内部类：SNode，TransferStack是由SNode单链表构建成的堆栈结构；只有一个head指针指向链表的表头；每次添加元素都是在表头处添加，新节点成为新的表头head；唤醒的线程的时候也是唤醒head节点，因此就形成先进后出的堆栈结构；TransferStack中根据e是否为null将线程分为2类；一类是获取元素：REQUEST,一类是添加元素：DATA;TransferStack与TransferQueue类似，TransferStack的堆栈中也只有一种节点，在栈中的节点被唤醒的时候才会短暂的出现2种节点。

![\`][945ea105e7eb492c93ac01777ed11cd6.png_pic_center]

在TransferStack的堆栈中，如果新加入的线程类型与堆栈中的节点类型不同，那么会先将新线程包装成SNode节点加入到堆栈中，成为新的head节点并将旧head节点唤醒；最后更新head节点返回DATA类型节点的元素值；

TransferStack的处理方法与TransferQueue中的处理不同，TransferQueue不会将新线程包装成节点，新线程会直接唤醒TransferQueue队列的head节点；

举个例子，假设堆栈中全是DATA节点，现在有一个take线程进入堆栈中：

![在这里插入图片描述][189532e74e2c4fbcaf1b79c63a122527.png_pic_center]

在有不同类型的节点进入堆栈中的时候，新节点添加到堆栈顶端并更新为新的head节点；这个节点的mode = REQUEST | FULFILLING ;FULFILLING 是用来标记，表示这个head节点正在唤醒堆栈中的一个节点线程；最后在新节点唤醒旧的head节点（ oldHead节点）之后，更新堆栈的head节点；

TransferStack部分的源码就再不分析了，入队阻塞部分的源码几乎与TransferQueue一样；TransferStack唤醒节点的方式与TransferQueue有点差别，TransferStack是将新节点先包装成节点添加到堆栈中，再唤醒节点线程，最后重新设置堆栈的head指针并将这2个节点清除出堆栈。TransferStack中还有其它很多细节，这些细节都很重要。不过在搞清楚了它入队出队的机制之后完全没有动力去思考这些了。其实这些细节是很重要的，正是这些细节保证了在没有使用锁的情况下让SynchronousQueue在多线程环境下使用仍然是线程安全的。暂时分析到这里，其他的细节想到再更新吧。

[e89822d6ed5f4fab8cbb12721808cbfb.png_pic_center]: https://img-blog.csdnimg.cn/e89822d6ed5f4fab8cbb12721808cbfb.png#pic_center
[3de779a32c8848d0b1903e7243680a7e.png_pic_center]: https://img-blog.csdnimg.cn/3de779a32c8848d0b1903e7243680a7e.png#pic_center
[c93d0a38e07b402a93acc5343507e32c.png_pic_center]: https://img-blog.csdnimg.cn/c93d0a38e07b402a93acc5343507e32c.png#pic_center
[355ed27baddc41b99f9895f6d58544ef.png_pic_center]: https://img-blog.csdnimg.cn/355ed27baddc41b99f9895f6d58544ef.png#pic_center
[9e4a7c3a337f4544aff77aafeb9a9a95.png_pic_center]: https://img-blog.csdnimg.cn/9e4a7c3a337f4544aff77aafeb9a9a95.png#pic_center
[61a7742026ac401295484a49aaba0c18.png_pic_center]: https://img-blog.csdnimg.cn/61a7742026ac401295484a49aaba0c18.png#pic_center
[945ea105e7eb492c93ac01777ed11cd6.png_pic_center]: https://img-blog.csdnimg.cn/945ea105e7eb492c93ac01777ed11cd6.png#pic_center
[189532e74e2c4fbcaf1b79c63a122527.png_pic_center]: https://img-blog.csdnimg.cn/189532e74e2c4fbcaf1b79c63a122527.png#pic_center