java.util.concurrent（JUC）包中的阻塞队列（全部）

淡淡的烟草味﹌ 2022-02-13 10:19 141阅读 0赞

### 文章目录 ###

*  0、java.util.concurrent包中各类的关系图
 *  1、阻塞队列
 *   *  1.1、BlockingQueue 接口
     *   *  1.1.1、BlockingQueue 用法
         *  1.1.2、 BlockingQueue 的方法
         *  1.1.3、BlockingQueue 的实现类
         *  1.1.4、 BlockingQueue 实现的生产者和消费者的例子
     *  1.2、数组队列 ArrayBlockingQueue
     *  1.3、延迟队列 DelayQueue
     *   *  1.3.1、使用 DelayQueue 的生产者和消费者的例子：
     *  1.4、链式的队列 LinkedBlockingQueue
     *  1.5、具有优先级的队列 PriorityBlockingQueue
     *   *  源码简要说明：
     *  1.6、同步队列 SynchronousQueue
 *  2、阻塞双端队列 BlockingDeque
 *   *  2.1、BlockingDeque 接口
     *   *  2.1.1、BlockingDeque 的使用
         *  2.1.2、BlockingDeque 的方法
         *  2.1.3、BlockingDeque 接口的方法2：
         *  2.1.4、BlockingDeque 的实现类
         *  2.1.5、BlockingDeque 代码示例
     *  2.2、链式的双向队列 LinkedBlockingDeque
 *  3、阻塞式转移队列 TransferQueue 接口
 *   *  3.1、链式的无界 LinkedTransferQueue （jdk1.7，推荐）
     *   *  3.1.1 设计原理

参考《Java 并发指南》  
http://tutorials.jenkov.com/java-util-concurrent/index.html  
https://blog.csdn.net/a724888/column/info/21961

# 0、java.util.concurrent包中各类的关系图 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9qaW4yMWNlbg_size_16_color_FFFFFF_t_70]

# 1、阻塞队列 #

## 1.1、BlockingQueue 接口 ##

### 1.1.1、BlockingQueue 用法 ###

BlockingQueue 通常用于一个线程去生产对象，另外一个线程去消费这些对象的场景。

原理示意图：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9qaW4yMWNlbg_size_16_color_FFFFFF_t_70 1]

一个线程在头部，往队列里边放元素；另一个线程在尾部，从队列中取的元素。

一个线程（生产者）将会持续生产新对象并将其插入到队列之中，直到队列达到它所能容纳的临界点。也就是说，它是有限的。如果该阻塞队列到达了其临界点，负责生产的线程将会在往里边插入新对象时发生阻塞。它会一直处于阻塞之中，直到负责消费的线程从队列中拿走一个对象。（**总结：生产者可以持续向队列插入新的元素，直到队列满为止，队列满后生产者线程被阻塞**）

负责消费的线程将会一直从该阻塞队列中拿出对象。如果消费线程尝试去从一个空的队列中提取对象的话，这个消费线程将会处于阻塞之中，直到一个生产线程把一个对象丢进队列。（**总结：消费者可以持续从队列取出元素，直到队列空为止，队列空后消费者线程被阻塞**）

### 1.1.2、 BlockingQueue 的方法 ###

BlockingQueue 具有 4 组不同的方法用于插入、移除以及对队列中的元素进行检查。这些方法如下：

<table> 
 <thead> 
  <tr> 
   <th>操作</th> 
   <th>抛异常 （boolean）</th> 
   <th>特定值 （boolean）</th> 
   <th>阻塞（void）</th> 
   <th>超时 （void）</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><strong>插入</strong></td> 
   <td>add(o)</td> 
   <td>offer(o)</td> 
   <td>put(o)</td> 
   <td>offer(o, timeout, timeunit)</td> 
  </tr> 
  <tr> 
   <td><strong>移除</strong></td> 
   <td>remove(o)</td> 
   <td>poll(o)</td> 
   <td>take(o)</td> 
   <td>poll(timeout, timeunit)</td> 
  </tr> 
  <tr> 
   <td><strong>检查</strong></td> 
   <td>element(o)</td> 
   <td>peek(o)</td> 
   <td></td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

四组不同的行为方式解释：

1.  **抛异常**：如果试图的操作无法立即执行，抛一个 `IllegalStateException` 异常；
2.  **特定值**：如果试图的操作无法立即执行，返回一个特定的值( true 或 false) ；
3.  **阻塞**：如果试图的操作无法立即执行，该方法调用将会发生阻塞，直到能够执行 ；
4.  **超时**：如果试图的操作无法立即执行，该方法调用将会发生阻塞，直到能够执行，但等待时间不会超过给定值。超出特定值则返回操作是否成功( true 或 false)。

如果你试图向 BlockingQueue 中插入一个 null，将会抛出一个 NullPointerException。

BlockingQueue 不适用于除头部，尾部外的元素遍历，元素遍历时效率很低 。另外，remove(o) 方法操作元素移除，效率并不高。

**注**：基于队列的数据结构，获取除开始或结束位置的其他对象的效率不会太高，因此你尽量不要用这一类的方法，除非你确实不得不那么做。

### 1.1.3、BlockingQueue 的实现类 ###

BlockingQueue 是个接口，它的实现类(Java 6)：

*  ArrayBlockingQueue
 *  DelayQueue
 *  LinkedBlockingQueue
 *  PriorityBlockingQueue
 *  SynchronousQueue
 *  LinkedTransferQueue （jdk1.7，推荐）

### 1.1.4、 BlockingQueue 实现的生产者和消费者的例子 ###

列子中，ArrayBlockingQueue 是一个有界队列，作为共享队列。

BlockingQueueExample 类分别在两个独立的线程中启动了一个 Producer 和 一个 Consumer。  
Producer 向一个共享的 BlockingQueue 中注入元素（字符串），而 Consumer 则从队列中取元素。

public class BlockingQueueExample {  
    
        public static void main(String[] args) throws Exception {  
    
            BlockingQueue queue = new ArrayBlockingQueue(1024); 
    
            Producer producer = new Producer(queue); 
            Consumer consumer = new Consumer(queue); 
    
            new Thread(producer).start(); 
            new Thread(consumer).start(); 
    
            Thread.sleep(4000); 
        } 
    }

Producer 类  
它在每次 put() 调用时是如何休眠一秒钟的，这将导致 Consumer 在等待队列中对象的时候发生阻塞。

public class Producer implements Runnable{ 
    
        protected BlockingQueue queue = null;
    
        public Producer(BlockingQueue queue) { 
            this.queue = queue;
        }
    
        public void run() { 
            try { 
                queue.put("1");
                Thread.sleep(1000);
                queue.put("2");
                Thread.sleep(1000);
                queue.put("3");
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }    
    }

Consumer 类  
它是把从队列中取出元素，然后打印。

public class Consumer implements Runnable { 
    
         protected BlockingQueue queue = null; 
    
         public Consumer(BlockingQueue queue) { 
               this.queue = queue; 
         }
    
         public void run() { 
               try {  
                     System.out.println(queue.take()); 
                     System.out.println(queue.take()); 
                     System.out.println(queue.take()); 
               } catch (InterruptedException e) { 
                     e.printStackTrace(); 
               } 
          } 
    }

## 1.2、数组队列 ArrayBlockingQueue ##

ArrayBlockingQueue 是一个 **有界** 的阻塞队列，其内部实现是将对象放到一个数组里。  
使用时，必须指定容量大小。因为它是基于数组实现的，也就具有数组的特性：一旦初始化后，则大小就无法修改。

ArrayBlockingQueue 内部以 FIFO(先进先出)的顺序对元素进行存储。队列中的头元素在所有元素之中是放入时间最久的那个，而尾元素则是最短的那个。

初始化的一个示例：

BlockingQueue queue = new ArrayBlockingQueue(1024); 
    
    queue.put("1"); 
    
    Object object = queue.take();

使用了 Java 泛型的示例，对 String 元素放入和提取的：

BlockingQueue<String> queue = new ArrayBlockingQueue<String>(1024); 
    
    queue.put("1"); 
    
    String string = queue.take();

## 1.3、延迟队列 DelayQueue ##

DelayQueue 对元素进行持有直到一个特定的延迟到期。

DelayQueue中的元素必须实现 java.util.concurrent.Delayed 接口，并重写下面两个方法：

/** * 获得延迟时间 */
     @Override
    public long getDelay(TimeUnit unit) { 
        return 0;
    }
    
    /** * 用于延迟队列内部比较排序 */
    @Override
    public int compareTo(Delayed o) { 
        return 0;
    }

**getDelay(TimeUnit unit)**

`getDelay()` 方法返回的值的时间段之后才释放掉该元素。如果返回的是 0 或者负值，延迟将被认为过期，该元素将会在 DelayQueue 的下一次 take 被调用的时候被释放掉。

getDelay 方法的 参数是一个枚举类型，它表明了将要延迟的时间段。TimeUnit 枚举将会取以下值：

DAYS
    HOURS
    MINUTES
    SECONDS
    MILLISECONDS
    MICROSECONDS
    NANOSECONDS

**compareTo(Delayed o)**

`compareTo(Delayed o)` 对 DelayQueue 队列中的元素进行排序时有用，因此它们可以根据过期时间进行有序释放。

### 1.3.1、使用 DelayQueue 的生产者和消费者的例子： ###

package com.aop8.queue.delay2;
    
    public class DelayQueueExample { 
    
    	public static void main(String[] args) { 
    		DelayQueue<DelayedElement> delayQueue = new DelayQueue<DelayedElement>();
    
    		LogPrint logPrint = new LogPrint(delayQueue);
    		Producer producer = new Producer(delayQueue);
    		Consumer consumer = new Consumer(delayQueue);
    
            //日志线程
    		new Thread(logPrint).start();
            //消费者线程
    		new Thread(consumer,"c1").start();
    		new Thread(consumer,"c2").start();
    		new Thread(consumer,"c3").start();
            //生产者线程
    		new Thread(producer,"p1").start();
    		new Thread(producer,"p2").start();
    
    		try { 
    			TimeUnit.MINUTES.sleep(60000 * 5);
    		} catch (InterruptedException e) { 
    			e.printStackTrace();
    		}
    	}
    }

DelayedElement 延迟队列中的元素  
延迟队列中的元素 必须实现 Delayed 接口，重写 getDelay() 和 compareTo() 方法。

package com.aop8.queue.delay2;
    
    public class DelayedElement implements Delayed { 
    
    	private final long delay; //延迟时间
        private final long expire;  //到期时间
        private final String msg;   //数据
        private final long now; //创建时间
    
        public DelayedElement(long delay, String msg) { 
            this.delay = delay;
            this.msg = msg;
            expire = System.currentTimeMillis() + delay;    //到期时间 = 当前时间+延迟时间
            now = System.currentTimeMillis();
        }
    
        /** * 需要实现的接口，获得延迟时间 用过期时间-当前时间 * @param unit * @return */
        @Override
        public long getDelay(TimeUnit unit) { 
            return unit.convert(this.expire - System.currentTimeMillis() , TimeUnit.MILLISECONDS);
        }
    
        /** * 用于延迟队列内部比较排序 当前时间的延迟时间 - 比较对象的延迟时间 * @param o * @return */
        @Override
        public int compareTo(Delayed o) { 
            return (int) (this.getDelay(TimeUnit.MILLISECONDS) -o.getDelay(TimeUnit.MILLISECONDS));
        }
    
        @Override
        public String toString() { 
            final StringBuilder sb = new StringBuilder("DelayedElement{");
            sb.append("delay=").append(delay);
            sb.append(", expire=").append(expire);
            sb.append(", msg='").append(msg).append('\'');
            sb.append(", now=").append(now);
            sb.append('}');
            return sb.toString();
        }
    }

LogPrint 日志打印类：

package com.aop8.queue.delay2;
    
    public class LogPrint implements Runnable { 
    
    	protected DelayQueue<DelayedElement> delayQueue = null;
    
    	public LogPrint(DelayQueue<DelayedElement> delayQueue) { 
    		this.delayQueue = delayQueue;
    	}
    
    	@Override
    	public void run() { 
    		while (true) { 
    			try { 
    				TimeUnit.MILLISECONDS.sleep(1000);
    			} catch (InterruptedException e) { 
    				e.printStackTrace();
    			}
    			System.out.println("当前队列元素个数：" + delayQueue.size());
    		}
    	}
    }

Producer 生产者：

package com.aop8.queue.delay2;
    
    public class Producer implements Runnable { 
    
    	protected DelayQueue<DelayedElement> delayQueue = null;
    
    	public Producer(DelayQueue<DelayedElement> delayQueue) { 
    		this.delayQueue = delayQueue;
    	}
    
    	public void run() { 
    		while (true) { 
    			try { 
    				
    				long random=new Random().nextInt(5)+1;
    				TimeUnit.MILLISECONDS.sleep(1000 * 2*random);
    			} catch (InterruptedException e) { 
    				e.printStackTrace();
    			}
    
    			long random=new Random().nextInt(5)+1;
    			DelayedElement element = new DelayedElement(500*random, "test");
    			boolean bb=delayQueue.offer(element);
    			if(bb){ 
    				System.out.println(Thread.currentThread().getName() + "--添加--" + element);
    			}
    		}
    	}
    }

消费者：

package com.aop8.queue.delay2;
    
    public class Consumer implements Runnable { 
    
    	protected DelayQueue<DelayedElement> delayQueue = null;
    
    	public Consumer(DelayQueue<DelayedElement> delayQueue) { 
    		this.delayQueue = delayQueue;
    	}
    
    	public void run() { 
    		while (true) { 
    			DelayedElement element = null;
    			try { 
    				element = delayQueue.take();
    			} catch (InterruptedException e) { 
    				e.printStackTrace();
    			}
    			System.out.println(Thread.currentThread().getName() + "--消费--" + element);
    		}
    	}
    }

## 1.4、链式的队列 LinkedBlockingQueue ##

LinkedBlockingQueue 内部以一个链式结构对其元素进行存储。如果需要的话，这一链式结构可以选择一个上限。如果没有定义上限，将使用 `Integer.MAX_VALUE` 作为上限。（**既是无界队列，又是有界队列**）

LinkedBlockingQueue 内部以 FIFO (先进先出) 的顺序对元素进行存储。队列中的头元素在所有元素之中是放入时间最久的那个，而尾元素则是最短的那个。

//创建一个容量是 Integer.MAX_VALUE 能力。 
    LinkedBlockingQueue() 
    
    //创建一个指定的大小的队列。 
    LinkedBlockingQueue(int capacity) 
    
    //创建一个容量大小是 Integer.MAX_VALUE 队列，最初包含给定集合的元素，添加到集合的迭代器遍历顺序。 
    LinkedBlockingQueue(Collection<? extends E> c)

示例代码：

BlockingQueue<String> unbounded = new LinkedBlockingQueue<String>();  //无界队列
    BlockingQueue<String> bounded = new LinkedBlockingQueue<String>(1024); //有界队列
    
    bounded.put("Value");
    
    String value = bounded.take();

参考文章：

https://blog.csdn.net/tonywu1992/article/details/83419448

## 1.5、具有优先级的队列 PriorityBlockingQueue ##

PriorityBlockingQueue 默认初始容量(11)，最大容量是 `int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8` 。

*  是一个无界有序的阻塞队列，排序规则和之前介绍的PriorityQueue一致，只是增加了阻塞操作。
 *  不支持插入null元素，
 *  PriorityBlockingQueue 队列中的元素 **必须** 实现 java.lang.Comparable 接口。
 *  它的迭代器并不保证队列保持任何特定的顺序，如果想要顺序遍历，考虑使用Arrays.sort(pq.toArray())。
 *  不保证 **同等优先级** 的元素顺序，如果你想要强制顺序，就需要考虑自定义顺序或者是Comparator使用第二个比较属性。

### 源码简要说明： ###

**数据结构**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9qaW4yMWNlbg_size_16_color_FFFFFF_t_70 2]

1.  DEFAULT\_INITIAL\_CAPACITY：默认队列容量是11 ；
2.  MAX\_ARRAY\_SIZE：最大可分配队列容量 `Integer.MAX_VALUE - 8`，减8是因为有的VM实现在数组头有些内容；
3.  queue：队列元素数组。平衡二叉堆实现，父节点下标是n，左节点则是2n+1，右节点是2n+2。最小的元素在最前面。
4.  size：当前队列中元素的个数 ；
5.  comparator：决定队列中元素先后顺序的比较器；
6.  lock：所有public方法的锁；
7.  notEmpty：队列为空时的阻塞条件；
8.  allocationSpinLock：扩容数组分配资源时的自旋锁，CAS需要；
9.  q：PriorityQueue只用于序列化的时候，为了兼容之前的版本。只有在序列化和反序列化的时候不为null。

**基本操作**

放入一个元素：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9qaW4yMWNlbg_size_16_color_FFFFFF_t_70 3]  
　和PriorityQueue的实现基本一致区别就是在于加锁了，并发出了非空信号唤醒阻塞的获取线程。  
　具体操作原理看：PriorityQueue。这里值得一说的就是tryGrow的实现，其用了一个while循环来处理，下面是具体实现。  
　![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9qaW4yMWNlbg_size_16_color_FFFFFF_t_70 4]  
　其先放开了锁，然后通过CAS设置allocationSpinLock来判断哪个线程获得了扩容权限，如果没抢到权限就会让出CPU使用权。最后还是要锁住开始真正的扩容。扩容权限争取到了就是计算大小，分配数组。暂不肯定为什么这么麻烦要分配数组的时候释放锁，暂猜测这样做效率会更高。

其它的操作过程都和PriorityQueue的类似，不再进行介绍。

**示例：**

BlockingQueue queue = new PriorityBlockingQueue();  //使用默认初始容量(11)
    
    //String implements java.lang.Comparable
    queue.put("Value");
    
    String value = queue.take();

参考文章：  
https://www.cnblogs.com/lighten/p/7510799.html

## 1.6、同步队列 SynchronousQueue ##

SynchronousQueue 是一个特殊的队列，它的内部同时 **只能够容纳单个元素** 。如果该队列中已有一个元素的话，试图向队列中插入一个新元素的线程将会阻塞，直到另一个线程将该元素从队列中抽走。同样，如果该队列为空，试图向队列中抽取一个元素的线程将会阻塞，直到另一个线程向队列中插入了一条新的元素。

据此，把这个类称作一个队列显然是夸大其词了。它更多像是一个汇合点。

**特点：**  
1、不能在同步队列上进行 peek，因为仅在试图要取得元素时，该元素才存在；  
2、除非另一个线程试图移除某个元素，否则也不能（使用任何方法）添加元素；也不能迭代队列，因为其中没有元素可用于迭代。队列的头是尝试添加到队列中的首个已排队线程元素； 如果没有已排队线程，则不添加元素并且头为 null。  
3、对于其他 Collection 方法（例如 contains），SynchronousQueue 作为一个空集合。此队列不允许 null 元素。  
4、它非常适合于传递性设计，在这种设计中，在一个线程中运行的对象要将某些信息、事件或任务传递给在另一个线程中运行的对象，它就必须与该对象同步。  
5、对于正在等待的生产者和使用者线程而言，此类支持可选的公平排序策略。默认情况下不保证这种排序。 但是，使用公平设置为 true 所构造的队列可保证线程以 FIFO 的顺序进行访问。 公平通常会降低吞吐量，但是可以减小可变性并避免得不到服务。

**方法：**

*  iterator() 永远返回空，因为里面没东西。
 *  peek() 永远返回null。
 *  put() 往queue放进去一个element以后就一直wait直到有其他thread进来把这个element取走。
 *  offer() 往queue里放一个element后立即返回，如果碰巧这个element被另一个thread取走了，offer方法返回true，认为offer成功；否则返回false。
 *  offer(2000, TimeUnit.SECONDS) 往queue里放一个element但是等待指定的时间后才返回，返回的逻辑和offer()方法一样。
 *  take() 取出并且remove掉queue里的element（认为是在queue里的。。。），取不到东西他会一直等。
 *  poll() 取出并且remove掉queue里的element（认为是在queue里的。。。），只有到碰巧另外一个线程正在往queue里offer数据或者put数据的时候，该方法才会取到东西。否则立即返回null。
 *  poll(2000, TimeUnit.SECONDS) 等待指定的时间然后取出并且remove掉queue里的element,其实就是再等其他的thread来往里塞。
 *  isEmpty()永远是true。
 *  remainingCapacity() 永远是0。
 *  remove()和removeAll() 永远是false。

SynchronousQueue 内部没有容量，但是由于一个插入操作总是对应一个移除操作，反过来同样需要满足。那么一个元素就不会再SynchronousQueue 里面长时间停留，一旦有了插入线程和移除线程，元素很快就从插入线程移交给移除线程。

也就是说这更像是一种信道（管道），资源从一个方向快速传递到另一方 向。显然这是一种**快速传递元素的方式**，也就是说在这种情况下元素总是以最快的方式从插入者（生产者）传递给移除者（消费者），这在多任务队列中是最快处理任务的方式。

在线程池里的一个**典型应用**是 `Executors.newCachedThreadPool()` 就使用了SynchronousQueue，这个线程池根据需要（新任务到来时）创建新的线程，如果有空闲线程则会重复使用，线程空闲了60秒后会被回收。

**示例：**

import java.util.concurrent.SynchronousQueue;
    
    public class SynchronousQueueDemo { 
        public static void main(String[] args) throws InterruptedException { 
            final SynchronousQueue<Integer> queue = new SynchronousQueue<Integer>();
    
            Thread putThread = new Thread(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println("put thread start");
                    try { 
                        queue.put(1);
                    } catch (InterruptedException e) { 
                    }
                    System.out.println("put thread end");
                }
            });
    
            Thread takeThread = new Thread(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println("take thread start");
                    try { 
                        System.out.println("take from putThread: " + queue.take());
                    } catch (InterruptedException e) { 
                    }
                    System.out.println("take thread end");
                }
            });
    
            putThread.start();
            Thread.sleep(1000);
            takeThread.start();
        }
    }

put thread start
    take thread start
    take from putThread: 1
    take thread end
    put thread end

# 2、阻塞双端队列 BlockingDeque #

BlockingDeque 接口表示一个线程安放入和提取实例的双端队列。队列满时，它将阻塞住试图插入元素的线程；队列为空时，它将阻塞住试图抽取的线程。

deque(双端队列) 是 “Double Ended Queue” 的缩写。因此，双端队列是一个你可以从任意一端插入或者抽取元素的队列。

## 2.1、BlockingDeque 接口 ##

### 2.1.1、BlockingDeque 的使用 ###

一个线程既是一个队列的生产者，又是这个队列的消费者。如果生产者线程需要在队列的两端都可以插入数据，消费者线程需要在队列的两端都可以移除数据，这时可以使用 BlockingDeque。

BlockingDeque 图解：  
![在这里插入图片描述][20190427185642166.png]

一个 BlockingDeque - 线程在双端队列的两端都可以插入和提取元素。

一个线程生产元素，并且把元素插入到队列的任意一端（头部 或 尾部 ）。  
如果双端队列已满，插入线程将被阻塞，直到一个移除线程从该队列中移出了一个元素。  
如果双端队列为空，移除线程将被阻塞，直到一个插入线程向该队列插入了一个新元素。

### 2.1.2、BlockingDeque 的方法 ###

BlockingDeque 具有 4 组不同的方法用于插入、移除以及对双端队列中的元素进行检查。如果请求的操作不能得到立即执行的话，每个方法的表现也不同。这些方法如下：

在头部提供了四种类型的操作方法：

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th>抛异常</th> 
   <th>特定值</th> 
   <th>阻塞</th> 
   <th>超时</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>插入</td> 
   <td>addFirst(o)</td> 
   <td>offerFirst(o)</td> 
   <td>putFirst(o)</td> 
   <td>offerFirst(o, timeout, timeunit)</td> 
  </tr> 
  <tr> 
   <td>移除</td> 
   <td>removeFirst(o)</td> 
   <td>pollFirst(o)</td> 
   <td>takeFirst(o)</td> 
   <td>pollFirst(timeout, timeunit)</td> 
  </tr> 
  <tr> 
   <td>检查</td> 
   <td>getFirst(o)</td> 
   <td>peekFirst(o)</td> 
   <td></td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

在尾部提供了四种类型的操作方法：

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th>抛异常</th> 
   <th>特定值</th> 
   <th>阻塞</th> 
   <th>超时</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>插入</td> 
   <td>addLast(o)</td> 
   <td>offerLast(o)</td> 
   <td>putLast(o)</td> 
   <td>offerLast(o, timeout, timeunit)</td> 
  </tr> 
  <tr> 
   <td>移除</td> 
   <td>removeLast(o)</td> 
   <td>pollLast(o)</td> 
   <td>takeLast(o)</td> 
   <td>pollLast(timeout, timeunit)</td> 
  </tr> 
  <tr> 
   <td>检查</td> 
   <td>getLast(o)</td> 
   <td>peekLast(o)</td> 
   <td></td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

四组不同的行为方式解释：

1.  **抛异常**：如果试图的操作无法立即执行，抛一个异常。
2.  **特定值**：如果试图的操作无法立即执行，返回一个特定的值(常常是 true / false)。
3.  **阻塞**：如果试图的操作无法立即执行，该方法调用将会发生阻塞，直到能够执行。
4.  **超时**：如果试图的操作无法立即执行，该方法调用将会发生阻塞，直到能够执行，但等待时间不会超过给定值。返回一个特定值以告知该操作是否成功(典型的是 true / false)。

### 2.1.3、BlockingDeque 接口的方法2： ###

BlockingDeque 接口继承自 BlockingQueue 接口。它拥用 BlockingQueue 的 所有方法，又可以在双端队列的头部和尾端进行操作 移除或添加新元素。

BlockingDeque 接口的方法：

<table> 
 <thead> 
  <tr> 
   <th>BlockingQueue</th> 
   <th>BlockingDeque</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><strong>添加</strong></td> 
   <td>-</td> 
  </tr> 
  <tr> 
   <td>add()</td> 
   <td>addLast()</td> 
  </tr> 
  <tr> 
   <td>offer() x 2</td> 
   <td>offerLast() x 2</td> 
  </tr> 
  <tr> 
   <td>put()</td> 
   <td>putLast()</td> 
  </tr> 
  <tr> 
   <td><strong>删除</strong></td> 
   <td>-</td> 
  </tr> 
  <tr> 
   <td>remove()</td> 
   <td>removeFirst()</td> 
  </tr> 
  <tr> 
   <td>poll() x 2</td> 
   <td>pollFirst()</td> 
  </tr> 
  <tr> 
   <td>take()</td> 
   <td>takeFirst()</td> 
  </tr> 
  <tr> 
   <td><strong>获取元素</strong></td> 
   <td>-</td> 
  </tr> 
  <tr> 
   <td>element()</td> 
   <td>getFirst()</td> 
  </tr> 
  <tr> 
   <td>peek()</td> 
   <td>peekFirst()</td> 
  </tr> 
 </tbody> 
</table>

### 2.1.4、BlockingDeque 的实现类 ###

既然 BlockingDeque 是一个接口，它的实现类：

*  LinkedBlockingDeque

### 2.1.5、BlockingDeque 代码示例 ###

代码示例：

BlockingDeque<String> deque = new LinkedBlockingDeque<String>();
    
    deque.addFirst("1");
    deque.addLast("2");
    
    String two = deque.takeLast();
    String one = deque.takeFirst();

## 2.2、链式的双向队列 LinkedBlockingDeque ##

LinkedBlockingDeque是基于双向链表实现的有界阻塞队列，默认大小为 `Integer.MAX_VALUE`，有较好的吞吐量，但可预测性差。

LinkedBlockingDeque 是一个由链表结构组成的双向阻塞队列，可以从队列的两端插入和移除元素。双向队列因为多了一个操作队列的入口，在多线程同时入队时，也就减少了一半的竞争。

相比于其他阻塞队列，LinkedBlockingDeque多了addFirst、addLast、peekFirst、peekLast等方法。

*  以 `first` 结尾的方法，表示对双端队列的第一个元素进行插入、获取获移除。
 *  以 `last` 结尾的方法，表示对双端队列的最后一个元素进行插入、获取获移除最后一个元素。

**LinkedBlockingDeque类有三个构造方法：**

public LinkedBlockingDeque() //默认容量大小为 Integer.MAX_VALUE
    public LinkedBlockingDeque(int capacity)  //指定容量大小
    public LinkedBlockingDeque(Collection<? extends E> c)

简单用法：

BlockingDeque<String> deque = new LinkedBlockingDeque<String>(); //无界队列
    BlockingDeque<String> deque2 = new LinkedBlockingDeque<String>(100); //指定capacity 大小时，就是有界队列
    
    deque.addFirst("1");
    deque.addLast("2");
    
    String two = deque.takeLast();
    String one = deque.takeFirst();

参考文章：  
https://blog.csdn.net/qq\_38293564/article/details/80592429

# 3、阻塞式转移队列 TransferQueue 接口 #

这个是JDK7才定义的一个节点，LinkedTransferQueue实现了这个接口，其新特性也就与之有关。

通常阻塞队里中，生产者放入元素，消费者使用元素，这两个部分是分离的。  
这里的分离意思如下：厨师做好了菜放在柜台上，服务员端走，厨师是不需要管有没有人取走做的菜，服务员也不需要管厨师有没有做好菜，没做好菜阻塞就行了。上面就是个人所说的分离的意思。

TransferQueue接口定义的相关内容就是厨师会知道做好的菜有没有被取走。

TransferQueue 接口继承了 BlockingQueue接口，因此具有BlockingQueue接口的所有方法，并增加了一些方法。  
![在这里插入图片描述][20190429133420397.png]

1.  tryTransfer(E)：将元素立刻给消费者。准确的说就是立刻给一个等待接收元素的线程，如果没有消费者就会返回false，而不将元素放入队列。
2.  transfer(E)：将元素给消费者，如果没有消费者就会等待。
3.  tryTransfer(E,long,TimeUnit)：将元素立刻给消费者，如果没有就等待指定时间。给失败返回false。
4.  hasWaitingConsumer()：返回当前是否有消费者在等待元素。
5.  getWaitingConsumerCount()：返回等待元素的消费者个数。

Java 8 提供了一个基于链表的实现类：

*  LinkedTransferQueue

## 3.1、链式的无界 LinkedTransferQueue （jdk1.7，推荐） ##

**起源：** 我觉得是这样的，之前的BlockingQueue是对 读取 或者 写入 锁定整个队列，所以在比较繁忙的时候，各种锁比较耗时

而当时有一个 SynchronizedQueue 其实不能叫Queue，因为只能放**一个物件**，要么有**一个物件**在等人拿，要么有**一个空**等人放

根据这个原理，诞生了 `LinkedTransferQueue`，利用CompareAndSwap进行一个无阻塞的队列，针对**每一个**操作进行处理样大家就不用抢得那么辛苦了 。

LinkedTransferQueue采用一种预占模式。意思就是消费者线程取元素时，如果队列不为空，则直接取走数据，若队列为空，那就生成一个节点（节点元素为null）入队，然后消费者线程被等待在这个节点上，后面生产者线程入队时发现有一个元素为null的节点，生产者线程就不入队了，直接就将元素填充到该节点，并唤醒该节点等待的线程，被唤醒的消费者线程取走元素，从调用的方法返回。我们称这种节点操作为“匹配”方式。

LinkedTransferQueue是 ConcurrentLinkedQueue（循环CAS+volatile 实现的wait-free并发算法）、SynchronousQueue（公平模式下转交元素）、LinkedBlockingQueue（阻塞Queue的基本方法）的超集。而且 LinkedTransferQueue更好用，因为它不仅仅综合了这几个类的功能，同时也提供了更高效的实现。

### 3.1.1 设计原理 ###

Dual Queues是该队列的基础理论。此队列不进存放数据节点，也会存放请求节点。当一个线程试图放入一个数据节点，正好遇到一个请求数据的结点，会立刻匹配并移除该数据节点，对于请求节点入队列也是一样的。Blocking Dual Queues阻塞所有未匹配的线程，直到有匹配的线程出现。

一个先入先出的dual queue实现是无锁队列算法M&S的变体。其包含两个指向字段：head指向一个匹配的结点，然后依次指向未匹配的结点，如果不为空。tail指向最后一个节点，或者null，如果队列为空。

例如下图是一个包含四个元素的队列结构：

![在这里插入图片描述][20190429133619834.png]

M&S算法易于扩展和保持(通过CAS)这些头部和尾指针。在dual队列中，节点需要自动维护匹配状态。所以这里需要一些必要的变量：对于数据模式，匹配需要将一个item字段通过CAS从非null的数据转成null，反之对于请求模式，需要从null变成data。一旦一个节点匹配了，其状态将不再改变。因此通常安排元素链表的前缀是0个或多个匹配节点，而后跟随0个或多个未匹配节点。如果不关心时间或空间的效率，通过从头指针开始遍历队列放入取出操作都是对的。CAS操作第一个未匹配节点匹配时的item，在下一个字段追加后一个节点。然而这是一个糟糕的想法，虽然其确实有好处，不需对head或tail进行原子更新。

LinkedTransferQueue采取了一种折中的方案，介于实时更新head/tail和不更新head/tail之间的方法。该方法对有时候需要额外的遍历去定位第一个或最后一个未匹配的结点和减少开销及队列结点的竞争更新这两个方面进行了权衡。

例如，一个可能出现的队列快照如下图：  
![在这里插入图片描述][20190429133626156.png]

slack（head位置和第一个未匹配的结点的最大距离，尾结点类似）的最佳值是一个经验问题，发现在1~3之间在大部分平台是最佳的值。更大的值会增加内存命中开销和长遍历链表的风险，更小的值则会增加CAS的竞争开销。

具体实现：使用一个基础的threshold来更新，slack为2。所以在当前位置超过第一个或最后一个节点2个距离以上的时候就会更新head/tail。出入队列操作都是通过xfer方法完成的，只需要不同的参数来表示操作。

https://www.cnblogs.com/lighten/p/7505355.html （含源码分析）  
https://www.jianshu.com/p/808da4a75f22  
https://blog.csdn.net/qq\_38293564/article/details/80593821

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9qaW4yMWNlbg_size_16_color_FFFFFF_t_70]: /images/20220213/b2402ff0488044edab1482266e18d166.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9qaW4yMWNlbg_size_16_color_FFFFFF_t_70 1]: /images/20220213/fb280b94e63541dbab92d4b5d1e8261d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9qaW4yMWNlbg_size_16_color_FFFFFF_t_70 2]: /images/20220213/1ecd235045254dd3a292e4e4003d6a5b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9qaW4yMWNlbg_size_16_color_FFFFFF_t_70 3]: /images/20220213/69d4a96f14874281a8838bc1873a73fc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hpYW9qaW4yMWNlbg_size_16_color_FFFFFF_t_70 4]: /images/20220213/0757a824f4b3474d8dad9f4abde3f804.png
[20190427185642166.png]: /images/20220213/1a29b39b44634d368d535f1e932a7cb9.png
[20190429133420397.png]: /images/20220213/864ddfb231f746cbb48517fd4e7d7003.png
[20190429133619834.png]: /images/20220213/6dd25cbb12ec4cb99a6952f0a6222e1d.png
[20190429133626156.png]: /images/20220213/cab307c18db943ef8a896a0fb8bda890.png