博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
阻塞队列---ArrayBlockingQueue,LinkedBlockingQueue,DelayQueue源码分析
阅读量:4925 次
发布时间:2019-06-11

本文共 15320 字,大约阅读时间需要 51 分钟。

阻塞队列和非阻塞队列

阻塞队列和非阻塞队列的区别:阻塞队列可以自己阻塞,非阻塞队列不能自己阻塞,只能使用队列wait(),notify()进行队列消息传送。而阻塞队列当队列里面没有值时,会阻塞直到有值输入。输入也一样,当队列满的时候,会阻塞,直到队列不为空。
阻塞队列不需要synchronized,或者调用wait,notify()来进行队列交互。

非阻塞队列:

queue.wait();queue.notify();是synchronized (queue)的queue对象调用的,睡眠和唤醒是线程的睡眠和唤醒,

public class feizusheQueue{    private int queueSize = 10;    private PriorityQueue
queue = new PriorityQueue
(queueSize); public static void main(String[] args) { feizusheQueue test = new feizusheQueue (); Producer producer = test.new Producer(); Consumer consumer = test.new Consumer(); producer.start(); consumer.start(); } class Consumer extends Thread{ @Override public void run() { consume(); } private void consume() { while(true){ synchronized (queue) { while(queue.size() == 0){ try { System.out.println("队列空,等待数据"); queue.wait(); } catch (InterruptedException e) { e.printStackTrace(); queue.notify(); } } queue.poll(); //每次移走队首元素 queue.notify();//执行完,不马上释放锁,synchronized执行完之后释放锁。 System.out.println("从队列取走一个元素,队列剩余"+ queue.size()+"个元素"); } } } } class Producer extends Thread{ @Override public void run() { produce(); } private void produce() { while(true){ synchronized (queue) { while(queue.size() == queueSize){ try { System.out.println("队列满,等待有空余空间"); queue.wait(); } catch (InterruptedException e) { e.printStackTrace(); queue.notify(); } } queue.offer(1); //每次插入一个元素 queue.notify(); System.out.println("向队列取中插入一个元素,队列剩余空间:"+ (queueSize-queue.size())); } } } }}

阻塞队列:

public class zusheQueue {    private int queueSize = 10;    private ArrayBlockingQueue
queue = new ArrayBlockingQueue
(queueSize); public static void main(String[] args) { zusheQueue test = new zusheQueue(); Producer producer = test.new Producer(); Consumer consumer = test.new Consumer(); producer.start(); consumer.start(); } class Consumer extends Thread{ @Override public void run() { consume(); } private void consume() { while(true){ try { queue.take(); System.out.println("从队列取走一个元素,队列剩余"+ queue.size()+"个元素"); } catch (InterruptedException e) { e.printStackTrace(); } } } } class Producer extends Thread{ @Override public void run() { produce(); } private void produce() { while(true){ try { queue.put(1); System.out.println("向队列取中插入一个元素,队列剩余空间:"+ (queueSize-queue.size())); } catch (InterruptedException e) { e.printStackTrace(); } } } }}

lock.lock();当其他线程获取了这个锁,这和线程也获取不到这个锁了。condition.signal();但是外层的lock没有释放,还是走不了的。只唤醒一个消费者。

 

阻塞队列中的几个主要方法:

put方法用来向队尾存入元素,如果队列满,则等待;
offer方法用来向队尾存入元素,如果队列满,则等待一定的时间,当时间期限达到时,如果还没有插入成功,则返回false;否则返回true;
  take方法用来从队首取元素,如果队列为空,则等待;
  poll方法用来从队首取元素,如果队列空,则等待一定的时间,当时间期限达到时,如果取到,则返回null;否则返回取得的元素;
put(E e) take() offer(E e,long timeout, TimeUnit unit) poll(long timeout, TimeUnit unit)

非阻塞队列中的几个主要方法:
add(E e):将元素e插入到队列末尾,如果插入成功,则返回true;如果插入失败(即队列已满),则会抛出异常;
remove():移除队首元素,若移除成功,则返回true;如果移除失败(队列为空),则会抛出异常;
offer(E e):将元素e插入到队列末尾,如果插入成功,则返回true;如果插入失败(即队列已满),则返回false;
poll():移除并获取队首元素,若成功,则返回队首元素;否则返回null;
peek():获取队首元素,若成功,则返回队首元素;否则返回null
对于非阻塞队列,一般情况下建议使用offer、poll和peek三个方法,不建议使用add和remove方法。因为使用offer、poll和peek三个方法可以通过返回值判断操作成功与否,而使用add和remove方法却不能达到这样的效果。注意,非阻塞队列中的方法都没有进行同步措施。

阻塞队列:
ArrayBlockingQueue:数组,在创建ArrayBlockingQueue对象时必须制定容量大小。并且可以指定公平性与非公平性,默认情况下为非公平的,即不保证等待时间最长的队列最优先能够访问队列
LinkedBlockingQueue:链表,在创建LinkedBlockingQueue对象时如果不指定容量大小,则默认大小为Integer.MAX_VALUE
PriorityBlockingQueue:以上2种队列都是先进先出队列,而PriorityBlockingQueue却不是,它会按照元素的优先级对元素进行排序,按照优先级顺序出队,每次出队的元素都是优先级最高的元素。注意,此阻塞队列为无界阻塞队列,即容量没有上限(通过源码就可以知道,它没有容器满的信号标志),前面2种都是有界队列。
DelayQueue:基于PriorityQueue,一种延时阻塞队列,DelayQueue中的元素只有当其指定的延迟时间到了,才能够从队列中获取到该元素。DelayQueue也是一个无界队列,因此往队列中插入数据的操作(生产者)永远不会被阻塞,而只有获取数据为空的操作(消费者)才会被阻塞。

public class ArrayBlockingQueue
extends AbstractQueue
implements BlockingQueue
, java.io.Serializable { final Object[] items; int takeIndex; //下一个要取的位置 int putIndex; //下一个要放的位置 int count; final ReentrantLock lock; private final Condition notEmpty; private final Condition notFull; public ArrayBlockingQueue(int capacity, boolean fair) { if (capacity <= 0) throw new IllegalArgumentException(); this.items = new Object[capacity]; lock = new ReentrantLock(fair); notEmpty = lock.newCondition(); notFull = lock.newCondition(); } public void put(E e) throws InterruptedException { checkNotNull(e); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == items.length) notFull.await(); enqueue(e); } finally { lock.unlock(); } } private void enqueue(E x) { // assert lock.getHoldCount() == 1; // assert items[putIndex] == null; final Object[] items = this.items; items[putIndex] = x; if (++putIndex == items.length) putIndex = 0; count++; notEmpty.signal(); } public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == 0) notEmpty.await(); return dequeue(); } finally { lock.unlock(); } } private E dequeue() { // assert lock.getHoldCount() == 1; // assert items[takeIndex] != null; final Object[] items = this.items; @SuppressWarnings("unchecked") E x = (E) items[takeIndex]; items[takeIndex] = null; if (++takeIndex == items.length) takeIndex = 0; count--; if (itrs != null) itrs.elementDequeued(); notFull.signal(); return x; } public boolean offer(E e) { checkNotNull(e); final ReentrantLock lock = this.lock; lock.lock(); try { if (count == items.length) return false; else { enqueue(e); return true; } } finally { lock.unlock(); } } public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException { checkNotNull(e); long nanos = unit.toNanos(timeout); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == items.length) { if (nanos <= 0)//时间到了就返回 return false; nanos = notFull.awaitNanos(nanos); } enqueue(e); return true; } finally { lock.unlock(); } } public E poll() { final ReentrantLock lock = this.lock; lock.lock(); try { return (count == 0) ? null : dequeue(); } finally { lock.unlock(); } } public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == 0) notEmpty.await(); return dequeue(); } finally { lock.unlock(); } } public E poll(long timeout, TimeUnit unit) throws InterruptedException { long nanos = unit.toNanos(timeout); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == 0) { if (nanos <= 0)//时间到了就返回 return null; nanos = notEmpty.awaitNanos(nanos); } return dequeue(); } finally { lock.unlock(); } } public E peek() { final ReentrantLock lock = this.lock; lock.lock(); try { return itemAt(takeIndex); // null when queue is empty } finally { lock.unlock(); } } public int size() { final ReentrantLock lock = this.lock; lock.lock(); try { return count; } finally { lock.unlock(); } }}

数组阻塞队列:一个lock,2个condition,放的时候放和取都不行,取的时候取和放都不行。为空了取的线程都在emptyCondition

等待,满了放的线程都在fullCondition上面等待,唤醒时候只唤醒一个线程。只能同时一个取或者放。

Condition的特性:

    1.Condition中的await()方法相当于Object的wait()方法,Condition中的signal()方法相当于Object的notify()方法,Condition中的signalAll()相当于Object的notifyAll()方法。不同的是,Object中的这些方法是和同步锁捆绑使用的;而Condition是需要与互斥锁/共享锁捆绑使用的。
    2.Condition它更强大的地方在于:能够更加精细的控制多线程的休眠与唤醒。对于同一个锁,我们可以创建多个Condition,在不同的情况下使用不同的Condition。
     如果采用Object类中的wait(), notify(), notifyAll()实现该缓冲区,当向缓冲区写入数据之后需要唤醒"读线程"时,不可能通过notify()或notifyAll()明确的指定唤醒"读线程",而只能通过notifyAll唤醒所有线程(但是notifyAll无法区分唤醒的线程是读线程,还是写线程)。 但是,通过Condition,就能明确的指定唤醒读线程

链表阻塞队列:2个锁,2个confition,放的时候不能放可以取(也只是一个取),取的时候不能取可以放(只能一个放)。只能同时一个放一个取。都是通过count来平衡空和满的。

public class LinkedBlockingQueue
extends AbstractQueue
implements BlockingQueue
, java.io.Serializable { static class Node
{ E item; Node
next; Node(E x) { item = x; } } private final int capacity; private final AtomicInteger count = new AtomicInteger(); transient Node
head; private transient Node
last; private final ReentrantLock takeLock = new ReentrantLock(); private final Condition notEmpty = takeLock.newCondition(); private final ReentrantLock putLock = new ReentrantLock(); private final Condition notFull = putLock.newCondition(); private void signalNotEmpty() { final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { notEmpty.signal(); } finally { takeLock.unlock(); } } private void signalNotFull() { final ReentrantLock putLock = this.putLock; putLock.lock(); try { notFull.signal(); } finally { putLock.unlock(); } } private void enqueue(Node
node) { last = last.next = node; } private E dequeue() { // assert takeLock.isHeldByCurrentThread(); // assert head.item == null; Node
h = head; Node
first = h.next; h.next = h; // help GC head = first; E x = first.item; first.item = null; return x; } void fullyLock() { putLock.lock(); takeLock.lock(); } void fullyUnlock() { takeLock.unlock(); putLock.unlock(); } public LinkedBlockingQueue(int capacity) { if (capacity <= 0) throw new IllegalArgumentException(); this.capacity = capacity; last = head = new Node
(null); } public int size() { return count.get(); } public int remainingCapacity() { return capacity - count.get(); } public void put(E e) throws InterruptedException { if (e == null) throw new NullPointerException(); int c = -1; Node
node = new Node
(e); final ReentrantLock putLock = this.putLock; final AtomicInteger count = this.count; putLock.lockInterruptibly(); try { while (count.get() == capacity) { notFull.await(); } enqueue(node); c = count.getAndIncrement(); if (c + 1 < capacity) notFull.signal(); } finally { putLock.unlock(); } if (c == 0) signalNotEmpty(); } public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException { if (e == null) throw new NullPointerException(); long nanos = unit.toNanos(timeout); int c = -1; final ReentrantLock putLock = this.putLock; final AtomicInteger count = this.count; putLock.lockInterruptibly(); try { while (count.get() == capacity) { if (nanos <= 0)//等待时间还没有就返回false return false; nanos = notFull.awaitNanos(nanos); } enqueue(new Node
(e)); c = count.getAndIncrement(); if (c + 1 < capacity) notFull.signal(); } finally { putLock.unlock(); } if (c == 0) signalNotEmpty(); return true; } public boolean offer(E e) { if (e == null) throw new NullPointerException(); final AtomicInteger count = this.count; if (count.get() == capacity) //是满的直接返回 return false; int c = -1; Node
node = new Node
(e); final ReentrantLock putLock = this.putLock; putLock.lock(); try { if (count.get() < capacity) { enqueue(node); c = count.getAndIncrement(); if (c + 1 < capacity) notFull.signal(); } } finally { putLock.unlock(); } if (c == 0) signalNotEmpty(); return c >= 0; } public E take() throws InterruptedException { E x; int c = -1; final AtomicInteger count = this.count; final ReentrantLock takeLock = this.takeLock; takeLock.lockInterruptibly(); try { while (count.get() == 0) { notEmpty.await(); } x = dequeue(); c = count.getAndDecrement(); if (c > 1) notEmpty.signal(); } finally { takeLock.unlock(); } if (c == capacity) signalNotFull(); return x; } public E poll(long timeout, TimeUnit unit) throws InterruptedException { E x = null; int c = -1; long nanos = unit.toNanos(timeout); final AtomicInteger count = this.count; final ReentrantLock takeLock = this.takeLock; takeLock.lockInterruptibly(); try { while (count.get() == 0) { if (nanos <= 0) //等到时间返回 return null; nanos = notEmpty.awaitNanos(nanos); } x = dequeue(); c = count.getAndDecrement(); if (c > 1) notEmpty.signal(); } finally { takeLock.unlock(); } if (c == capacity) signalNotFull(); return x; } public E poll() { final AtomicInteger count = this.count; if (count.get() == 0) return null; //直接返回 E x = null; int c = -1; final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { if (count.get() > 0) { x = dequeue(); c = count.getAndDecrement(); if (c > 1) notEmpty.signal(); } } finally { takeLock.unlock(); } if (c == capacity) signalNotFull(); return x; } public E peek() { if (count.get() == 0) return null; final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { Node
first = head.next; if (first == null) return null; else return first.item; } finally { takeLock.unlock(); } } public boolean remove(Object o) { if (o == null) return false; fullyLock();//读写都锁住 try { for (Node
trail = head, p = trail.next; p != null; trail = p, p = p.next) { if (o.equals(p.item)) { unlink(p, trail); return true; } } return false; } finally { fullyUnlock(); } } public boolean contains(Object o) { if (o == null) return false; fullyLock();//读写都锁住 try { for (Node
p = head.next; p != null; p = p.next) if (o.equals(p.item)) return true; return false; } finally { fullyUnlock(); } } public Object[] toArray() { fullyLock(); try { int size = count.get(); Object[] a = new Object[size]; int k = 0; for (Node
p = head.next; p != null; p = p.next) a[k++] = p.item; return a; } finally { fullyUnlock(); } }

 

转载于:https://www.cnblogs.com/yaowen/p/10695338.html

你可能感兴趣的文章
【t016】邮递员
查看>>
boost安装
查看>>
Vue与React的异同
查看>>
360:跳高游戏
查看>>
CSS3 Background-size
查看>>
Python Ethical Hacking - MAC Address & How to Change(3)
查看>>
生成验证码
查看>>
深入理解计算机系统 第2章 信息的表示和处理
查看>>
JS中数据结构之链表
查看>>
Tomcat 配置文件 (server.xml)详解--转载
查看>>
Golang理解-字符串拼接的几种方式
查看>>
Linux系统-RPM库顺坏
查看>>
Golang理解-错误处理策略
查看>>
运维职责
查看>>
Golang理解-集合
查看>>
element中使用表单验证rules 需要注意
查看>>
AnyChatSDK 实现视频通话
查看>>
Gerrit error when Change-Id in commit messages are missing
查看>>
【面试题】实现一个队列数据结构,并使用这个队列实现一个生产者消费者模式...
查看>>
第二阶段团队冲刺第二天站立会议
查看>>