为了加强公司网站建设,中国外贸网站排名,佛山市建设企业网站服务机构,泉港网站建设推广服务公司一、阻塞队列介绍
1、队列 队列入队从队首开始添加#xff0c;直至队尾#xff1b;出队从队首出队#xff0c;直至队尾#xff0c;所以入队和出队的顺序是一样的
Queue接口 add(E) #xff1a;在指定队列容量条件下添加元素#xff0c;若成功返回true#xff0c;若当前…一、阻塞队列介绍
1、队列 队列入队从队首开始添加直至队尾出队从队首出队直至队尾所以入队和出队的顺序是一样的
Queue接口 add(E) 在指定队列容量条件下添加元素若成功返回true若当前队列没有可用空间抛出IllegalStateException异常offer(E)在指定队列容量条件下添加元素若成功返回true若当前队列没有可用空间返回falseremove()返回并删除此队列的头部元素若队列为空会抛出异常poll()返回并删除此队列的头部元素若队列为空会返回nullelement()返回头部元素但不删除队列为空会抛出异常peek()返回头部元素但不删除队列为空返回null
2、阻塞队列
BlockingQueue规范定义了添加和删除阻塞队列的方法很多阻塞队列都是基于BlockingQueue实现的具体原理当阻塞队列插入数据时如果队列已满线程会阻塞等待直到队列非满从阻塞队列取数据时如果队列已空线程会阻塞等待直到队列非空
1BlockingQueue接口 put()将指定元素插入队列如果必要等待队列空间变为可用take()返回并删除队列中的头部元素如果必要直到等待某个元素可用offer(E, long, TimeUnit)将指定的元素插入此队列指定的等待时间等待必要的可用空间poll(long, TimeUnit)返回并删除此队列的头部元素指定的等待时间直到等待某个元素可用
2应用场景
线程池线程池中线程创建的个数超过核心线程数会放入到等待队列中如果队列空了核心线程又没有要处理的任务会进入等待直到队列中有新的任务生产者-消费者模式当生产者线程发现队列满了会陷入等待直到有消费者线程进行消费并唤醒生产者线程当消费者线程发现队列中没有可处理消息会陷入等待直到生产者线程进行生产并唤醒消费者线程阻塞队列可以避免线程间的竞争消息队列可以把消息放到队列中进行消息的异步处理缓存系统使用contains()方法判断是否包含某个元素利用阻塞队列来缓存数据避免多线程更新缓存的竞争并发任务处理将任务提交到队列中消费之后出队避免重复消费
3、JUC包下的阻塞队列 二、ArrayBlockingQueue
ArrayBlockingQueue采用Object数组方式存储数据创建ArrayBlockingQueue必须指定容量大小属于有界队列采用ReentrantLock保证线程安全如果生产速度和消费速度基本匹配的情况下使用ArrayBlockingQueue是个不错选择
1、使用 public class ArrayBlockingQueueTest {private static final int QUEUE_CAPACITY 5;private static final int PRODUCER_DELAY_MS 1000;private static final int CONSUMER_DELAY_MS 2000;public static void main(String[] args) throws InterruptedException {// 创建一个容量为QUEUE_CAPACITY的阻塞队列BlockingQueueString queue new ArrayBlockingQueue(QUEUE_CAPACITY);// 创建一个生产者线程Runnable producer () - {while (true) {try {// 在队列满时阻塞queue.put(producer);System.out.println(生产了一个元素队列中元素个数 queue.size());Thread.sleep(PRODUCER_DELAY_MS);} catch (InterruptedException e) {e.printStackTrace();}}};new Thread(producer).start();// 创建一个消费者线程Runnable consumer () - {while (true) {try {// 在队列为空时阻塞String element queue.take();System.out.println(消费了一个元素队列中元素个数 queue.size());Thread.sleep(CONSUMER_DELAY_MS);} catch (InterruptedException e) {e.printStackTrace();}}};new Thread(consumer).start();}
} 生产者少休眠1s生产的快当生产者添加第六个元素时会陷入等待
2、源码分析 items数组元素数组takeIndex下一个待取出元素索引putIndex下一个待添加元素索引count元素个数lock内置锁notEmpty消费者notFull生产者
入队详解
https://www.processon.com/view/link/64c8c537b9f7806c73dadbb4
出队详解
https://www.processon.com/view/link/64c8c92fb9f7806c73daea85
为什么ArrayBlockingQueue对数组操作要设计成双指针
如果用一个指针对数组的删除或者添加操作数组中的元素都要往前或者往后移动这样导致时间复杂度为O(n)而使用双指针可以前移后移可以提升操作的性能时间复杂度为O(1)
三、LinkedBlockingQueue
LinkedBlockingQueue是基于链表实现的阻塞队列队列默认大小为Integer.MAX_VALUE由于这个数值比较大LinkedBlockingQueue也被称为无界队列LinkedBlockingQueue每个元素都会占用内存为防止OOM还是设置一个队列大小
1、使用
和ArrayBlockingQueue使用基本差不多 LinkedBlockingQueue()队列大小为2的32次方减1LinkedBlockingQueue(Collection? extends E)队列大小为2的32次方减1按照传入集合初始化队列数据LinkedBlockingQueue(int)传入参数指定队列大小
2、源码分析
相比ArrayBlockingQueue读写只一把独占锁的实现LinkedBlockingQueue读写分了两把锁 item元素存储的数据next下一个节点单项链表结构capacity队列容量count元素数量head链表表头last链表表尾takeLock出队操作竞争的锁对象notEmpty当队列无元素时会让进行takeLock的线程陷入等待直到有线程唤醒putLock入队操作竞争的锁对象notFull当队列满了会让进行putLock的线程陷入等待直到有线程唤醒 初始化LinkedBlockingQueue对象时会创建一个属性item为null的Node对象
入队详解
https://www.processon.com/view/link/64c8f6d5b9f7806c73db4fc9
出队详解
https://www.processon.com/view/link/64c8ff0e7807695f1493090f
3、LinkedBlockingQueue和ArrayBlockingQueue对比
队列大小ArrayBlockingQueue必须指定容量大小LinkedBlockingQueue可以不指定LinkedBlockingQueue如果添加比删除快会导致OOM数组存储容器不同ArrayBlockingQueue采用数组存储数据LinkedBlockingQueue采用对象链表方式存储数据就因为会产生Node对象并发量大时会对gc产生较大的影响ArrayBlockingQueue添加和删除都是争抢同一个锁资源LinkedBlockingQueue添加和删除进行了锁分离LinkedBlockingQueue高并发场景下可以并行的进行入队和出队操作
四、DelayQueue
可以使用队列消息延迟消费实现接口回调通知、token超时失效、订单超时失效
1、使用
public class DelayQueueTest {public static void main(String[] args) throws InterruptedException {DelayQueueOrder delayQueue new DelayQueue();delayQueue.put(new Order(order1, System.currentTimeMillis(), 5000));delayQueue.put(new Order(order2, System.currentTimeMillis(), 2000));delayQueue.put(new Order(order3, System.currentTimeMillis(), 3000));while (!delayQueue.isEmpty()) {Order take delayQueue.take();System.out.println(处理订单take.getOrderId());}}static class Order implements Delayed {private String orderId;private long createTime;private long delayTime;public Order(String orderId, long createTime, long delayTime) {this.orderId orderId;this.createTime createTime;this.delayTime delayTime;}public String getOrderId() {return orderId;}Overridepublic long getDelay(TimeUnit unit) {// 订单创建时间延迟时间-当前时间剩余延迟时间long diff createTime delayTime - System.currentTimeMillis();return unit.convert(diff, unit);}Overridepublic int compareTo(Delayed o) {// 比较两个订单之间差多长时间long diff this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);return Long.compare(diff, 0);}}
}
2、源码分析 lock用于保证线程安全q 优先级队列,存储元素用于保证延迟低的优先执行leader用于标记当前是否有线程在排队仅用于取元素时 leader 指向的是第一个从队列获取元素阻塞的线程available条件用于表示现在是否有可取的元素 当新元素到达或新线程可能需要成为leader时被通知
入队详解
https://www.processon.com/view/link/64c91879470d721c4e3be985
出队详解
https://www.processon.com/view/link/64c9185fc1af4746895281e7
五、如何选择适合的阻塞队列
1、选择策略
通常我们可以从以下 5 个角度考虑来选择合适的阻塞队列
功能
第 1 个需要考虑的就是功能层面比如是否需要阻塞队列帮我们排序如优先级排序、延迟执行等。如果有这个需要我们就必须选择类似于 PriorityBlockingQueue 之类的有排序能力的阻塞队列。
容量
第 2 个需要考虑的是容量或者说是否有存储的要求还是只需要“直接传递”。在考虑这一点的时候我们知道前面介绍的那几种阻塞队列有的是容量固定的如 ArrayBlockingQueue有的默认是容量无限的如 LinkedBlockingQueue而有的里面没有任何容量如 SynchronousQueue而对于 DelayQueue 而言它的容量固定就是 Integer.MAX_VALUE。所以不同阻塞队列的容量是千差万别的我们需要根据任务数量来推算出合适的容量从而去选取合适的 BlockingQueue。
能否扩容
第 3 个需要考虑的是能否扩容。因为有时我们并不能在初始的时候很好的准确估计队列的大小因为业务可能有高峰期、低谷期。如果一开始就固定一个容量可能无法应对所有的情况也是不合适的有可能需要动态扩容。如果我们需要动态扩容的话那么就不能选择 ArrayBlockingQueue 因为它的容量在创建时就确定了无法扩容。相反PriorityBlockingQueue 即使在指定了初始容量之后后续如果有需要也可以自动扩容。所以我们可以根据是否需要扩容来选取合适的队列。
内存结构
第 4 个需要考虑的点就是内存结构。我们分析过 ArrayBlockingQueue 的源码看到了它的内部结构是“数组”的形式。和它不同的是LinkedBlockingQueue 的内部是用链表实现的所以这里就需要我们考虑到ArrayBlockingQueue 没有链表所需要的“节点”空间利用率更高。所以如果我们对性能有要求可以从内存的结构角度去考虑这个问题。
性能
第 5 点就是从性能的角度去考虑。比如 LinkedBlockingQueue 由于拥有两把锁它的操作粒度更细在并发程度高的时候相对于只有一把锁的 ArrayBlockingQueue 性能会更好。另外SynchronousQueue 性能往往优于其他实现因为它只需要“直接传递”而不需要存储的过程。如果我们的场景需要直接传递的话可以优先考虑 SynchronousQueue。
2、线程池对于阻塞队列的选择
线程池有很多种不同种类的线程池会根据自己的特点来选择适合自己的阻塞队列。
Executors类下的线程池类型
FixedThreadPoolSingleThreadExecutor 同理选取的是 LinkedBlockingQueueCachedThreadPool 选取的是 SynchronousQueueScheduledThreadPoolSingleThreadScheduledExecutor同理选取的是延迟队列
