首页 > 其他分享> > 阻塞队列实现及方法辨析

阻塞队列实现及方法辨析

2021-10-22 12:31:22 作者：互联网

阻塞队列

队列
阻塞队列
常用阻塞队列

队列

队列是一种特殊的线性表，特殊之处在于它只允许在表的前端（front）进行删除操作，而在表的后端（reat）进行插入操作，和栈一样，队列是一种操作受限制的线性表，进行插入操作的端称为尾端，进行删除操作的端称为队头
在队列中插入一个队列元素的动作称为入队，从队列中删除一个队列元素称为出队，因为队列只允许在一端插入，在另一端删除，所以只有最早进入队列的元素才能最先从队列中删除，故队列又称为先进先出（FIFO–first in first out）线性表

阻塞队列

支持阻塞的插入方法：意思是当队列满了的时候，队列会阻塞插入元素的线程，知道队列不为满的状态
支持阻塞的移除方法，意思是在队列为空时，获取元素的线程会等待队列变为非空

在并发编程中使用生产者和消费者模式能解决绝大多数并发问题，该模式通过平衡生产线程和消费线程的工作能力来提高程序整体处理数据的速度
在线程中，生产者即生产数据的线程，消费者即消费数据的线程，在多线程开发中，如果生产者处理速度大于消费者速度，那么生产者就必须等待消费者处理完数据才能继续生产数据，如果消费者的处理能力大于生产者，那消费者就必须等待生产者
为解决生产消费能力不均匀的问题，便提出了生产者和消费者模式，生产者和消费者模式是通过一个容器来解决生产者和消费者的 强耦合 问题。生产者和消费者彼此之间不直接通信，而是通过阻塞队列来进行通信，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里拿到数据，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力
阻塞队列常用于生产者和消费者的场景，生产者就是向队列添加元素的线程，消费者是从队列里面取出元素的线程，阻塞队列就是生产者用了来存放元素、消费者用来获取元素的容器

处理方式 \ 方法	插入方法	移除方法	检查方法
抛出异常	add(e)	remove()	element()
返回特殊值	offer(e)	poll()	peek()
持续阻塞	put(e)	take()	无
超时退出	offer(e, time, unit)	poll(time, unit)	无

抛出异常：当队列满时，如果再往队列插入元素，会抛出 illegalStateExcepion (“Queuefull”) 异常，当队列空的时候，从队列里获得元素会抛出 NoSuchElementException 异常
返回特殊值：当往队列插入元素时，会返回元素是否插入成功，成功返回 ture ，如果时移除方法，则是从队列中取出一个元素，如果没有则返回 null
持续阻塞：当阻塞队列满的时候，如果生产者线程往队列中 put 元素，队列会一直阻塞生产线程，知道队列可用或者响应中断退出，当队列空的时候，如果消费者线程从队列中 take 元素，队列会阻塞时住消费者线程，直到队列不为空
超时退出：当阻塞队列满的时候，如果生产者线程往队列里插入元素，队列会阻塞生产者一段时间，如果超过了指定时间，生产者线程就会退出

/***类说明：取出到期的订单的功能*/
public class FetchOrder implements Runnable {
	private DelayQueue<ItemVo<Order>> queue;

    public FetchOrder(DelayQueue<ItemVo<Order>> queue){
        this.queue = queue;

常用阻塞队列

ArrayBlockingQueue：一个由数组结构组成的有界阻塞队列
LinkedBlockingQueue：一个由链表结构组成的有界阻塞队列
PriorityBlockingQueue：一个支持优先级排序的无界阻塞队列
DelayQueue：一个使用优先级队列实现的无界阻塞队列
SynchronousQueue：一个不存储元素的阻塞队列
LinkedTransferQueue：一个由链表结构组成的无界阻塞队列
LinkedBlockingDeque：一个由链表结构组成的双向阻塞队列

以上的阻塞队列都实现了BlockingQueue接口，也都是线程安全的

有界阻塞 && 无界阻塞

有限队列即长度有限，满了以后生产者会进入阻塞状态，无界队列就是里面能放无限的东西而不会因为队列长度限制被阻塞，当然空间限制来源于系统资源的限制，如果处理不及时，导致队列越来越大，超出一定的限制致使内存超限，操作系统或者JVM就会直接将 OOM 杀死
需要注意：无界也会产生阻塞，因为阻塞不仅仅体现在生产者放入元素时会阻塞，消费者那元素时时，如果没有元素，同样也会阻塞

ArrayBlockingQueue

ArrayBlockingQueue 是一个用数组实现的有界阻塞队列，此队列按照先进先出（FIFO）的原则对元素进行排序，默认情况下不保证线程间可以公平地访问队列，所谓公平访问队列就是值阻塞的线程可以按照阻塞的先后顺序访问队列，即先阻塞的线程先访问队列，非公平性是对先等待的线程是非公平的，当队列可用时，阻塞的线程都可以争夺访问队列的资格，有可能先阻塞的线程最后才访问队列，初始化时有参数可以设置

LinkedBlockingQueue

LinkedBlockingQueue是一个用链表实现的有界阻塞队列，此队列的默认和最大长度为 Inreger.MAX_VALUE，此队列按照先进先出的原则对元素进行排序

Array 实现和 Linked 实现地区别

队列中锁的实现不同

ArrayBlockingQueue 实现的队列中的锁是没有分离的，即生产和消费用的是同一个锁
LinkedBlockingQueue实现的队列中的锁是分离的，即生产用的是putLock, 消费是 TakeLock

在生产或消费时操作不同

ArrayBlockingQueue实现的队列中在生产和消费的时候，是直接将枚举对象插入或移除的
LinkedBlockingQueue实现的队列中在生产和消费的时候，需要把枚举对象转换为Node 进行插入或移除，会影响性能

列队大小初始化方式不同

ArrayBlockingQueue实现的队列中必须指定队列的大小
LinkedBlockingQueue实现的队列中可以不指定列队的大小，但是默认是Integer.MAX_VALUE

PriorityBlockingQueue

PriorityBlockingQueue是一个支持优先级的无界阻塞队列。默认情况下元素采取自然顺序升序排列。也可以自定义类实现 compareTo() 方法（比较器）来指定元素排序规则，或者初始化 PriorityBlockingQueue 时，指定构造参数 Comparator 来对元素进行排序。需要注意的是不能保证同优先级元素的顺序。

DelayQueue

DelayQueue是一个支持延迟获取元素的无界阻塞队列，队列使用 PriorityQueue 来实现，队列中的元素必须实现 Delayed 接口，在创建元素时可以指定多久才能从队列中获取当前元素，只有在延迟期到期时才能从队列中提取元素

应用场景：
单机缓存系统的设计：可以用 DelayQueue 保存缓存元素的有效期，使用一个线程循环查询DelayQueue，一旦能从 DelayQueue 阻塞列队获取元素时，表示该元素有效期到了

实验代码：

类说明：取出到期的订单的功能：

/**
 *类说明：取出到期的订单的功能
 */
public class FetchOrder implements Runnable {
	private DelayQueue<ItemVo<Order>> queue;
	
    public FetchOrder(DelayQueue<ItemVo<Order>> queue){
        this.queue = queue;
    }

	@Override
	public void run() {
		while(true) {
			try {
				ItemVo<Order> item = queue.take();
				Order order = (Order)item.getData();
				System.out.println("Get From Queue:"+"data=" +order.getOrderNo()+";"+order.getOrderMoney());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		
	}	
}

类说明：存放的队列的元素:

import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

/**
 *类说明：存放的队列的元素，
 */
public class ItemVo<T> implements Delayed {

	//到期时间,但传入的数值代表过期的时长，传入单位毫秒
	private long activeTime;
	private T data;//业务数据，泛型

	//传入过期时长,单位秒，内部转换
	public ItemVo(long expirationTime, T data) {
		this.activeTime = expirationTime*1000+System.currentTimeMillis();
		this.data = data;
	}

	public long getActiveTime() {
		return activeTime;
	}

	public T getData() {
		return data;
	}

	/*
	 * 这个方法返回到激活日期的剩余时间，时间单位由单位参数指定。
	 */
	public long getDelay(TimeUnit unit) {
		long d = unit.convert(this.activeTime
				-System.currentTimeMillis(),unit);
		return d;
	}

	/*
	 *Delayed接口继承了Comparable接口，按剩余时间排序，实际计算考虑精度为纳秒数
	 */
	public int compareTo(Delayed o) {
		long d = (getDelay(TimeUnit.MILLISECONDS)
				- o.getDelay(TimeUnit.MILLISECONDS));
		if (d==0){
			return 0;
		}else{
			if (d<0){
				return -1;
			}else{
				return  1;
			}
		}
	}

}

类说明：订单的实体类:

/**
 *类说明：订单的实体类
 */
public class Order {
	private final String orderNo;//订单的编号
	private final double orderMoney;//订单的金额
	
	public Order(String orderNo, double orderMoney) {
		super();
		this.orderNo = orderNo;
		this.orderMoney = orderMoney;
	}

	public String getOrderNo() {
		return orderNo;
	}

	public double getOrderMoney() {
		return orderMoney;
	}
	
	
	
}

类说明：将订单推入队列：

import java.util.concurrent.DelayQueue;

/**
 *类说明：将订单推入队列
 */
public class PutOrder implements Runnable {
    private DelayQueue<ItemVo<Order>> queue;

    public PutOrder(DelayQueue<ItemVo<Order>> queue){
        this.queue = queue;
    }

	@Override
	public void run() {
		//5秒后到期
		Order orderTb = new Order("Tb12345",366);
		ItemVo<Order> itemTb = new ItemVo<Order>(5,orderTb);
		queue.offer(itemTb);
		System.out.println("订单5秒后超时："+orderTb.getOrderNo()+";"
                +orderTb.getOrderMoney());
		//8秒后到期
		Order orderJd = new Order("Jd54321",366);
		ItemVo<Order> itemJd = new ItemVo<Order>(8,orderJd);
		queue.offer(itemJd);
		System.out.println("订单8秒后超时："+orderJd.getOrderNo()+";"
                +orderJd.getOrderMoney());

	}	
}

类说明：延时队列测试程序：

import java.util.concurrent.DelayQueue;

/**
 *类说明：延时队列测试程序
 */
public class Test {
    public static void main(String[] args) throws InterruptedException {
        DelayQueue<ItemVo<Order>> queue = new DelayQueue<ItemVo<Order>>();//延时队列

        new Thread(new PutOrder(queue)).start();
        new Thread(new FetchOrder(queue)).start();

        //每隔500毫秒，打印个数字
        for(int i=1;i<15;i++){
            Thread.sleep(500);
            System.out.println(i*500);
        }
    }
}

SynchronousQueue

SynchronousQueue 是一个不存储元素的阻塞队列。每一个 put 操作必须等待一个 take 操作（put和take是持续输出的方法），否则不能继续添加元素。SynchronousQueue 可以看成是一个传球手，负责把生产者线程处理的数据直接传递给消费者线程。队列本身并不存储任何元素，非常适合传递性场景。SynchronousQueue 的吞吐量高于LinkedBlockingQueue 和 ArrayBlockingQueue

LinkedTransferQueue

LinkedTrransgerQueue主要方法：

transfer()

如果在当前有消费者正在等待接收元素（消费者使用take()方法或带时间限制的poll()方法时），transfer() 可以把生产者传入的元素立刻传输给消费者，如果没有消费者在等待接收元素，transfer() 会将元素存放到队列的 tail 节点，等到该元素被消费者消费了才返回

tryTransfer()

tryTransfer() 是用来试探生产者或传入的元素是否能直接传给消费者，如果没有消费者等待接收元素，则返回 false,和transfer() 的区别就是tryTransfer() 无论消费者是否接受，方法立刻返回，而 transfer() 是必须等到消费者消费了才返回

LinkedBlockingDeque

LinkedBlockingDeque 是一个由链表结构组成的双向阻塞队列。所谓双向队列指的是可以从队列的两端插入和移出元素。双向队列因为多了一个操作队列的入口，在多线程同时入队时，也就减少了一半的竞争。
多了addFirst、addLast、offerFirst、offerLast、peekFirst 和 peekLast 等方法，以First单词结尾的方法，表示插入、获取（peek）或移除双端队列的第一个元素。以Last单词结尾的方法，表示插入、获取或移除双端队列的最后一个元素。另外，插入方法add等同于addLast，移除方法 remove 等效于removeFirst。但是 take 方法却等同于 takeFirst，不知道是不是JDK的bug，使用时还是用带有First和Last后缀的方法更清楚。在初始化LinkedBlockingDeque 时可以设置容量防止其过度膨胀。另外，双向阻塞队列可以运用在“工作窃取”模式中

标签：队列,辨析,元素,阻塞,queue,线程,public
来源： https://blog.csdn.net/weixin_44766232/article/details/120894325