|
| 1 | +# BlockingQueue |
| 2 | +看一下官方1.8的官方文档 |
| 3 | + |
| 4 | + |
| 5 | +官方文档告诉我们它们有这些特点: |
| 6 | +- BlockingQueue实现被设计为主要用于生产者 - 消费者队列 |
| 7 | +- BlockingQueue实现是线程安全的。 所有排队方法使用内部锁或其他形式的并发控制在原子上实现其效果。 |
| 8 | +- BlockingQueue方法有四种形式，具有不同的操作方式，不能立即满足，但可能在将来的某个时间点满足： |
| 9 | + - 一个抛出异常 |
| 10 | + - 返回一个特殊值（ null或false ，具体取决于操作） |
| 11 | + - 第三个程序将无限期地阻止当前线程，直到操作成功为止 |
| 12 | + - 在放弃之前只有给定的最大时限。 |
| 13 | + |
| 14 | +## 看源码之旅 |
| 15 | + |
| 16 | +### ArrayBlockingQueue |
| 17 | +> 看一下官方文档的解释->一个有限的blocking queue由数组支持。 这个队列排列元素FIFO（先进先出）。 队列的头部是队列中最长的元素。 队列的尾部是队列中最短时间的元素。 新元素插入队列的尾部，队列检索操作获取队列头部的元素。这是一个经典的“有界缓冲区”，其中固定大小的数组保存由生产者插入的元素并由消费者提取。 创建后，容量无法更改。 尝试put成满的队列的元件将导致在操作阻挡; 尝试take从空队列的元件将类似地阻塞。此类支持可选的公平策略，用于订购等待的生产者和消费者线程。 默认情况下，此订单不能保证。 然而，以公平设置为true的队列以FIFO顺序授予线程访问权限。 公平性通常会降低吞吐量，但会降低变异性并避免饥饿。 |
| 18 | +
|
| 19 | +#### 常见变量 |
| 20 | +```java |
| 21 | +int count; // 队列元素数量 |
| 22 | +final ReentrantLock lock; // 内部线程安全性用了ReentrantLock |
| 23 | +private final Condition notEmpty; // takes方法的等待条件 |
| 24 | +private final Condition notFull; // puts方法的等待条件 |
| 25 | +``` |
| 26 | +#### 构造方法 |
| 27 | +```java |
| 28 | +public ArrayBlockingQueue(int capacity, boolean fair) { |
| 29 | + if (capacity <= 0) |
| 30 | + throw new IllegalArgumentException(); // 这里就不说了 |
| 31 | + this.items = new Object[capacity]; // 初始容量 |
| 32 | + lock = new ReentrantLock(fair); // fair参数决定是否公平锁 |
| 33 | + notEmpty = lock.newCondition(); // 上面已提到 |
| 34 | + notFull = lock.newCondition(); // 上面已提到 |
| 35 | +} |
| 36 | +``` |
| 37 | +#### add |
| 38 | +```java |
| 39 | +public boolean add(E e) { |
| 40 | + return super.add(e); // 该方法添加失败抛出异常 |
| 41 | +} |
| 42 | +// AbstractQueue |
| 43 | +public boolean add(E e) { |
| 44 | + if (offer(e)) |
| 45 | + return true; |
| 46 | + else |
| 47 | + throw new IllegalStateException("Queue full"); // 在这 |
| 48 | +} |
| 49 | +``` |
| 50 | +**remove就不分析了** |
| 51 | + |
| 52 | +#### offer |
| 53 | +```java |
| 54 | +public boolean offer(E e) { |
| 55 | + checkNotNull(e); |
| 56 | + final ReentrantLock lock = this.lock; // 上面的add其实内部也调用了offer，当时我还觉得奇怪，add没上锁？。 原来offer上了锁的 |
| 57 | + lock.lock(); |
| 58 | + try { |
| 59 | + if (count == items.length) |
| 60 | + return false; // 满了，就false |
| 61 | + else { |
| 62 | + enqueue(e); // 否则，添加即可 |
| 63 | + return true; // 返回true，并不会抛出异常 |
| 64 | + } |
| 65 | + } finally { |
| 66 | + lock.unlock(); |
| 67 | + } |
| 68 | +} |
| 69 | +``` |
| 70 | +**poll不分析了** |
| 71 | + |
| 72 | +#### put |
| 73 | +```java |
| 74 | +public void put(E e) throws InterruptedException { |
| 75 | + checkNotNull(e); // 检查是否为空，为空就抛出空异常 |
| 76 | + final ReentrantLock lock = this.lock; // 上锁哦 |
| 77 | + lock.lockInterruptibly(); // 锁中断 |
| 78 | + try { |
| 79 | + while (count == items.length) // 满了，挂起阻塞 |
| 80 | + notFull.await(); |
| 81 | + enqueue(e); // 否则添加 |
| 82 | + } finally { |
| 83 | + lock.unlock(); |
| 84 | + } |
| 85 | +} |
| 86 | +``` |
| 87 | + |
| 88 | +#### take |
| 89 | +```java |
| 90 | +public E take() throws InterruptedException { |
| 91 | + final ReentrantLock lock = this.lock; // 上锁 |
| 92 | + lock.lockInterruptibly(); // 锁中断 |
| 93 | + try { |
| 94 | + while (count == 0) |
| 95 | + notEmpty.await(); // 没有元素，挂起 |
| 96 | + return dequeue(); |
| 97 | + } finally { |
| 98 | + lock.unlock(); |
| 99 | + } |
| 100 | +} |
| 101 | +``` |
| 102 | + |
| 103 | +#### 带时间的offer和poll |
| 104 | +> 其实就是用了`long nanos = unit.toNanos(timeout);` |
| 105 | +
|
| 106 | + |
| 107 | +### LinkedBlockingQueue |
| 108 | +> 四个方法和ArrayBlockingQueue的差不多，就不多分析了。它的特点之一在于，如果不指定容量，那么默认是等于Integer.MAX_VALUE。可以看一下源码 |
| 109 | +
|
| 110 | +#### 常见的参数 |
| 111 | +```java |
| 112 | +/** The capacity bound, or Integer.MAX_VALUE if none */ |
| 113 | +private final int capacity; // 一看注释，就晓得了 |
| 114 | +// 还有一些常见的和上面的差不多 |
| 115 | +``` |
| 116 | + |
| 117 | +#### 构造方法 |
| 118 | +```java |
| 119 | +public LinkedBlockingQueue() { |
| 120 | + this(Integer.MAX_VALUE); // 在这里， 嘿嘿嘿。 |
| 121 | +} |
| 122 | +``` |
| 123 | + |
| 124 | +### LinkedTransferQueue |
| 125 | +> 基于链接节点的无界TransferQueue 。 这个队列相对于任何给定的生产者订购元素FIFO（先进先出）。 队列的头部是那些已经排队的元素是一些生产者的最长时间。 队列的尾部是那些已经在队列上的元素是一些生产者的最短时间。 |
| 126 | +
|
| 127 | +**四个方法就暂时不提了，大部分操作用了cas并且要关注transfer方法** |
| 128 | + |
| 129 | +先介绍几个标志参数： |
| 130 | +```java |
| 131 | +private static final int NOW = 0; // for untimed poll, tryTransfer |
| 132 | +private static final int ASYNC = 1; // for offer, put, add |
| 133 | +private static final int SYNC = 2; // for transfer, take |
| 134 | +private static final int TIMED = 3; // for timed poll, tryTransfer |
| 135 | +``` |
| 136 | + |
| 137 | +```java |
| 138 | +// 如果可能，立即将元素转移到等待的消费者。 |
| 139 | +public boolean tryTransfer(E e) { |
| 140 | + return xfer(e, true, NOW, 0) == null; // xfer的now参数 |
| 141 | +} |
| 142 | +// 还有一个重载参数的，带时间的。 |
| 143 | +``` |
| 144 | + |
| 145 | +```java |
| 146 | +// 将元素传输到消费者，必要时等待。 |
| 147 | +public void transfer(E e) throws InterruptedException { |
| 148 | + if (xfer(e, true, SYNC, 0) != null) { // SYNC |
| 149 | + Thread.interrupted(); // failure possible only due to interrupt |
| 150 | + throw new InterruptedException(); |
| 151 | + } |
| 152 | +} |
| 153 | +``` |
| 154 | +### PriorityBlockingQueue |
| 155 | +> 一看名字就是优先级阻塞队列，它的优先级是由堆实现的，所以该类中有很多堆的方法。源码暂时就不看了，很多都是差不多的。 |
| 156 | +
|
| 157 | +### SynchronousQueue |
| 158 | +> 官方文档：每个插入操作必须等待另一个线程相应的删除操作，反之亦然。 同步队列没有任何内部容量，甚至没有一个容量。 个人感觉是生产者生产一个元素，消费者必须消费，生产者才能继续生产。内部也维护了一个TransferQueue，其中部分操作是利用cas。源码就不贴了。有兴趣的可以进去看看哦。 |
0 commit comments