【重难点】【JUC 03】怎么实现一个线程安全的队列、手写模拟实现一个阻塞队列

文章目录

【重难点】【JUC 03】怎么实现一个线程安全的队列、手写模拟实现一个阻塞队列
一、怎么实现一个线程安全的队列
- 1.阻塞队列
- 2.非阻塞队列
- 3.总结
二、手写模拟实现一个阻塞队列
- 1.使用 synchronized 实现
- 2.使用 ReentrantLock

一、怎么实现一个线程安全的队列

Java 提供的线程安全的队列可以分为阻塞队列和非阻塞队列，其中阻塞队列的典型例子是 BlockingQueue，非阻塞队列的典型例子是 ConcurrentLinkedQueue，在实际应用中要根据实际需要选用阻塞队列或者非阻塞队列

1.阻塞队列

顾名思义，可以提供阻塞功能的队列

阻塞队列提供的常用方法：

	可能报异常	可能阻塞	返回布尔值	设定等待时间
入队	add(e)	offer(e)	put(e)	offer(e, timeout, unit )
出队	remove()	poll()	take()	poll(timeout, unit)
查看	element()	peek()	无	无

add、remove、element 方法不会阻塞线程，当不满足约束条件时，会抛出 IllegalStateException 异常。例如：当队列满时调用 add
offer、poll、peek 方法既不会阻塞线程，也不会抛出异常。例：当队列满时调用 offer，则不会插入元素，会返回 false
要想实现阻塞功能，就要调用 put 和 take 方法

ArrayBlockingQueue

基于数组的阻塞队列，在 ArrayBlocking 内部，维护了一个定长数组，以便缓存队列中的数据对象。此外，内部还保存着两个整形变量，分别标识着队列的头部和尾部在数组中的位置

ArrayBlockingQueue 在生产者放入数据和消费者获取数据时，用的是同一个锁对象，这也意味着两者无法真正并行运行，这一点尤其不同于 LinkedBlockingQueue。按照实现原理来分析，ArrayBlockingQueue 完全可以采用分离锁，从而实现生产者和消费者完全并行运行。之所以不这么做是因为 ArrayBlockingQueue 的数据写入和获取已经足够轻巧，以至于引入分离锁，除了给代码带来不必要的复杂性外，在性能上没有任何优势。ArrayBlockingQueue 和 LinkedBlockingQueue 间还有一个明显的不同之处在于：前者在插入或删除元素时不会产生或销毁任何额外的对象实例，而后者则会生成一个额外的 Node 对象。这在长时间内需要高效并发地处理大批量数据的系统中，其对于 GC 还是有一定的影响

LinkedBlockingQueue

基于链表的阻塞队列，与 ArrayListBlockingQueue 类似，其内部也维护了一个数据缓冲队列，区别在于这个队列由链表构成。当生产者向队列中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者立即返回。只有当队列缓冲区达到最大值缓存容量时（LinkedBlcokingQueue 可以通过构造函数指定该值），才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生产者线程会被唤醒，对于消费者的处理也是如此。LinkedBlockingQueue 之所以能够高效地处理并发数据，还有一个原因，其对于生产者和消费者采用了分离锁来控制数据同步。需要注意的是，如果创建了一个 LinkedBlockingQueue 对象而没有指定容量，LinkedBlockingQueue 会默认设置为 Integer.MAX_VALUE。这样就会带来一个问题，如果生产者的生产速度远大于消费者的消费速度，可能会导致 OOM

2.非阻塞队列

当许多线程共享访问一个公共集合时，ConcurrentLinkedQueue 是一个恰当的选择，此队列不允许 null 元素

3.总结

在并发编程中，一般推荐使用阻塞队列，这样可以尽量避免程序出现意外的错误。阻塞队列最经典的应用场景就是 socket 客户端数据的读取和解析，读取数据的线程不断地将数据放入队列，然后解析线程不断地从队列获取数据解析

使用非阻塞队列虽然可以即使返回结果或者消费结果，但是必须自行编码解决返回为空的情况和消费重试等问题

它们都是线程安全的，不用考虑线程同步问题

二、手写模拟实现一个阻塞队列

1.使用 synchronized 实现

由于 synchronized 是同一把锁，所以使用 notify() 可能会唤醒非目标线程，notifyAll() 唤醒全部线程则会带来大量的 CPU 上下文交换和锁竞争

public class ArrayBlockingQueue{private Object[] array;      //数组private int head;           //头private int tail;            //尾private volatile int size;   //元素个数public ArrayBlockingQueue(int capacity){this.array = new Object[capacity];}//写入元素public synchronized void put(Object o) throws InterruptedException{//当队列满时，阻塞while(size == array.length){this.awit();}array[tail++] = o;if(tail ==array.length){tail = 0;}size++;//唤醒线程this.notifyAll();}//取出元素public synchronized Object get() throws InterruptedException{//当队列为空，阻塞while(size == 0){this.wait();}Object o = array[head++];if(head == array.length){head = 0;}size--;//唤醒线程this.notifyAll();return o;}
}

2.使用 ReentrantLock

可以使用 Condition 指定要唤醒的线程，所以效率高

public class ArrayBlockingQueue{private Object[] array;      //数组private int head;           //头private int tail;            //尾private volatile int size;   //元素个数private ReentrantLock lock = new ReentrantLock();    //锁private Condition notEnpty = lock.newCondition();   //非空private Condition notFull = lock.newCondition();   //非满public ArrayBlockingQueue(int capacity){this.array = new Object[capacity];}//写入元素public void put(Object o) throws InterruptedException{try{lock.lock();//当队列满时，阻塞while(size == array.length){notFull.wait();}array[tail++] = o;if(tail == array.length){tail = 0;}size++;//唤醒线程notEmpty.notifyAll();}finally{lock.unlock();}}//取出元素public Object get() throws InterruptedException{lock.lock();try{//当队列为空，阻塞while(size == 0){notEmpty.wait();}Object o = array[head++];if(head == array.length){head = 0;}size--;//唤醒线程notFull.notyfyAll();return o;}finally{lock.unlock();}}
}