Java-并发-SynchronousQueue

小鱼儿 2024-04-18 09:01 62阅读 0赞

## 转载声明 ##

***本文大量内容系转载自以下文章，有删改，并参考其他文档资料加入了一些内容：***

*  [SynchronousQueue的使用][SynchronousQueue]  
    作者：零度anngle
 *  [TransferQueue 以及 SynchronousQueue][TransferQueue _ SynchronousQueue]  
    作者：薇薇一笑g

***转载仅为方便学习查看，一切权利属于原作者，本人只是做了整理和排版，如果带来不便请联系我删除。***

## 摘要 ##

在研究ThreadPool源码的时候，看到`Executors.newScheduledThreadPool`用了SynchronousQueue做接收task的队列，之前对SynchronousQueue也不了解，所以写下此文。

## 1 SynchronousQueue基础概念 ##

### 1.1 概述 ###

SynchronousQueue也是一种线程安全的阻塞队列，线程每次调用`put`方法必须等待另一个线程`take`方法，反之亦然。即其中每个插入操作必须等待另一个线程的移除操作，同样任何一个移除操作都等待另一个线程的插入操作。SynchronousQueue没有任何内部容量，因为不会真正存放元素。

SynchronousQueue非常适合于传递性设计，在这种设计中，在一个线程中运行的对象要将某些信息、事件或任务传递给在另一个线程中运行的对象，它就必须与该对象同步。

### 1.2 关于公平性 ###

SynchronousQueue默认非公平，但对于生产者/消费者模型线程可使用公平排序策略可保证线程以 FIFO 的顺序进行访问。 公平策略通常会降低吞吐量，但是可以减小可变性并避免某些线程得不到服务。 比如生产者消费者模型(生产者生产速度大于消费者消费速度)中，如果采用非公平策略，多个生产者中的某些生产者生产的物品永远不会被消费！而采用公平策略，则会机会均等的被消费。

### 1.3 重要方法 ###

SynchronousQueue的一些特殊点如下：

*  不允许使用`null`元素
 *  `iterator()` 永远返回空，即不能迭代队列，因为其中没有元素可用于迭代。队列的头是尝试添加到队列中的首个已排队线程元素； 如果没有已排队线程，则不添加元素并且头为 null。
 *  对于其他 Collection 方法（例如 `contains, containsAll, removeAll, retainAll`），SynchronousQueue 作为一个空集合，都会返回false，而`isEmpty()`永远是true，`remainingCapacity()` 永远是0。
 *  某线程`put()` 往queue放进去一个element以后，一直wait直到有其他线程进来把这个element取走。
 *  关于`offer()`：除非另一个线程试图移除某个元素，否则也不能（使用任何方法）添加元素;
    
    `offer()` 往queue里放一个element后立即返回，如果碰巧这个element被另一个thread取走了，offer方法返回true，认为offer成功；否则返回false。
    
    `offer(2000, TimeUnit.SECONDS)` 往queue里放一个element但是等待指定的时间后才返回结果，返回值的判定逻辑和`offer()`方法一样。
 *  `peek()`永远返回null，因为peek的含义是仅拿出头元素而不删除。但SynchronousQueue的内涵就是移除时才可能有元素。
 *  `take()`尝试取出element，取不到就一直阻塞等待。
 *  `poll()` 取出element，仅当碰巧另外一个线程正在往queue里放数据的时候，poll方法才会取到东西并返回；否则立即返回`null`
 *  `poll(2000, TimeUnit.SECONDS)` 等待指定的时间然后取出element，其实就是在等其他线程放入数据。

## 2 SynchronousQueue存在意义 ##

### 2.1 概述 ###

一个没有容量的并发队列有什么用了？或者说存在的意义是什么？

SynchronousQueue 的实现非常复杂，内部没有容量，但是由于一个插入操作总是对应一个移除操作，反过来同样需要满足。那么一个元素就不会再SynchronousQueue 里面长时间停留，一旦有了插入线程和移除线程，元素很快就从插入线程移交给移除线程了。也就是说这更像是一种管道，资源从一个方向快速传递到另一方向。

需要特别说明的是，尽管元素在SynchronousQueue 内部不会停留，但是并不意味之SynchronousQueue 内部没有队列。实际上SynchronousQueue 维护者**线程队列**，也就是插入线程或者移除线程在不同时存在的时候就会有阻塞的线程队列。既然有队列，同样就有公平性和非公平性特性，公平性保证正在等待的插入线程/移除线程以FIFO的顺序传递资源。

显然这是一种快速传递元素的方式，也就是说在这种情况下元素总是以最快的方式从插入着（生产者）传递给移除着（消费者），这在多任务队列中是最快处理任务的方式。

SynchronousQueue模拟的功能类似于生活中一手交钱一手交货这种情形，像那种货到付款或者先付款后发货模型不适合使用SynchronousQueue。首先要知道SynchronousQueue没有容纳元素的能力，即它的isEmpty()方法总是返回true，但是给人的感觉却像是只能容纳一个元素。

### 2.2 生产者-消费者例子 ###

下面使用SynchronousQueue模拟只能生产一个产品的生产者-消费者模型。代码如下：

import java.util.Random;
    import java.util.concurrent.SynchronousQueue;
    import java.util.concurrent.TimeUnit;
     
    public class SynchronousQueueTest { 
    	public static void main(String[] args) throws InterruptedException { 
    		SynchronousQueue<Integer> queue = new SynchronousQueue<Integer>();
    	
    		new Product(queue).start();
    		new Customer(queue).start();
    	}
    	static class Product extends Thread{ 
    		SynchronousQueue<Integer> queue;
    		public Product(SynchronousQueue<Integer> queue){ 
    			this.queue = queue;
    		}
    		@Override
    		public void run(){ 
    			while(true){ 
    				int rand = new Random().nextInt(1000);
    				System.out.println("生产了一个产品："+rand);
    				System.out.println("等待三秒后运送出去...");
    				try { 
    					TimeUnit.SECONDS.sleep(3);
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				}
    				try { 
    					queue.put(rand);
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				}
    				
    				System.out.println(queue.isEmpty());
    			}
    		}
    	}
    	static class Customer extends Thread{ 
    		SynchronousQueue<Integer> queue;
    		public Customer(SynchronousQueue<Integer> queue){ 
    			this.queue = queue;
    		}
    		@Override
    		public void run(){ 
    			while(true){ 
    				try { 
    					System.out.println("消费了一个产品:"+queue.take());
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				}
    				System.out.println("------------------------------------------");
    			}
    		}
    	}
    }

运行结果：

生产了一个产品：542
    等待三秒后运送出去...
    true
    消费了一个产品:542
    生产了一个产品：183
    等待三秒后运送出去...
    ------------------------------------------
    true
    消费了一个产品:183
    ------------------------------------------
    生产了一个产品：583
    等待三秒后运送出去...
    true
    消费了一个产品:583
    ------------------------------------------

### 2.3 Executors.newCachedThreadPool ###

该问题请参考[Executors.newCachedThreadPool][]

## 3 TransferQueue ##

### 3.1 概述 ###

`TransferQueue`继承了BlockingQueue（BlockingQueue又继承了Queue）并扩展了一些新方法。

*  BlockingQueue（和Queue）是Java 5中加入的接口，它是指这样的一个队列：当生产者向队列添加元素但队列已满时，生产者会被阻塞；当消费者从队列移除元素但队列为空时，消费者会被阻塞。
 *  `TransferQueue`则更进一步，生产者会一直阻塞直到所添加到队列的元素被某一个消费者所消费（不仅仅是添加到队列里就完事）。新添加的transfer方法用来实现这种约束。顾名思义，阻塞就是发生在元素从一个线程transfer到另一个线程的过程中，它有效地实现了元素在线程之间的传递（以建立Java内存模型中的happens-before关系的方式）。

TransferQueue还包括了其他的一些方法：两个`tryTransfer`方法，一个是非阻塞的，另一个带有timeout参数设置超时时间的。还有两个辅助方法`hasWaitingConsumer()`和`getWaitingConsumerCount()`。

### 3.2 TransferQueue对比SynchronousQueue ###

SynchronousQueue的队列长度为0，最初我认为这好像没多大用处，但后来我发现它是整个Java Collection Framework中最有用的队列实现类之一，**特别是对于两个线程之间传递元素**这种用例。

TransferQueue相比SynchronousQueue用处更广、更好用，因为你可以决定是使用BlockingQueue的方法（译者注：例如put方法）还是确保一次传递完成（译者注：即transfer方法）。在队列中已有元素的情况下，调用transfer方法，可以确保队列中被传递元素之前的所有元素都能被处理。

Doug Lea说从功能角度来讲，`LinkedTransferQueue`实际上是`ConcurrentLinkedQueue`、`SynchronousQueue`（公平模式）和`LinkedBlockingQueue`的超集。而且LinkedTransferQueue更好用，因为它不仅仅综合了这几个类的功能，同时也提供了更高效的实现。

Joe Bowbeer提供了一篇[William Scherer, Doug Lea, and Michael Scott][William Scherer_ Doug Lea_ and Michael Scott]的论文，在这篇论文中展示了LinkedTransferQueue的算法，性能测试的结果表明它优于Java 5的那些类（译者注：ConcurrentLinkedQueue、SynchronousQueue和LinkedBlockingQueue）。LinkedTransferQueue的性能分别是SynchronousQueue的3倍（非公平模式）和14倍（公平模式）。因为像ThreadPoolExecutor这样的类在任务传递时都是使用SynchronousQueue，所以使用LinkedTransferQueue来代替SynchronousQueue也会使得ThreadPoolExecutor得到相应的性能提升。考虑到executor在并发编程中的重要性，你就会理解添加这个实现类的重要性了。

Java 5中的SynchronousQueue使用两个队列（一个用于正在等待的生产者、另一个用于正在等待的消费者）和一个用来保护两个队列的锁。而LinkedTransferQueue使用CAS操作（译者注：参考wiki）实现一个非阻塞的方法，这是避免序列化处理任务的关键。这篇论文还罗列了很多的细节和数据，如果你感兴趣，非常值得一读。

## 更多好文 ##

*  [线程池队列 SynchronousQueue 详解][SynchronousQueue 1]
 *  [同步队列SynchronousQueue 生产消费另一个例子][SynchronousQueue 2]

[SynchronousQueue]: https://blog.csdn.net/zmx729618/article/details/52980158
[TransferQueue _ SynchronousQueue]: https://www.cnblogs.com/guweiwei/p/7123274.html
[Executors.newCachedThreadPool]: https://blog.csdn.net/baichoufei90/article/details/87167708#CachedThreadPool
[William Scherer_ Doug Lea_ and Michael Scott]: http://www.cs.rice.edu/~wns1/papers/2006-PPoPP-SQ.pdf
[SynchronousQueue 1]: https://blog.csdn.net/weixin_41622183/article/details/89283085
[SynchronousQueue 2]: https://www.cnblogs.com/qjm201000/p/10146359.html