J.U.C 提供的阻塞队列
在 Java8 中，提供了 7 个阻塞队列
-
ArrayBlockingQueue
数组实现的有界阻塞队列, 此队列按照先进先出（FIFO）的原则对元素进行排序。
-
LinkedBlockingQueue
链表实现的有界阻塞队列, 此队列的默认和最大长度为。Integer.MAX_VALUE。此队列按照先进先出的原则对元素进行排序。
-
PriorityBlockingQueue
支持优先级排序的无界阻塞队列, 默认情况下元素采取自然顺序升序排列。也可以自定义类实现 compareTo()方法来指定元素排序规则，或者初始化 - PriorityBlockingQueue 时，指定构造参数 Comparator 来对元素进行排序。
-
DelayQueue
优先级队列实现的无界阻塞队列。
-
SynchronousQueue
不存储元素的阻塞队列, 每一个 put 操作必须等待一个 take 操作，否则不能继续添加元素。
-
LinkedTransferQueue
链表实现的无界阻塞队列。
-
LinkedBlockingDeque
链表实现的双向阻塞队列。
阻塞队列的操作方法
在阻塞队列中，为插入和移除操作各提供了四种处理方式。
-
插入操作
- add(e)
添加元素到队列中，如果队列满了，继续插入元素会报错，IllegalStateException。 - offer(e)
添加元素到队列，同时会返回元素是否插入成功的状态，如果成功则返回 true。 - put(e)
当阻塞队列满了以后，生产者继续通过 put添加元素，队列会一直阻塞生产者线程，知道队列可用。 - offer(e,time,unit)
当阻塞队列满了以后继续添加元素，生产者线程会被阻塞指定时间，如果超时，则线程直接退出。
- add(e)
-
移除操作
remove()
当队列为空时，调用 remove 会返回 false，
如果元素移除成功，则返回 true。poll():
当队列中存在元素，则从队列中取出一个元素，
如果队列为空，则直接返回 null。take()：
基于阻塞的方式获取队列中的元素，如果队列为空，则 take 方法会一直阻塞，直到队列中有新的数据可以消费。poll(time,unit)：
带超时机制的获取数据，如果队列为空，则会等待指定的时间再去获取元素返回。
ArrayBlockingQueue 原理
-
构造方法
ArrayBlockingQueue 提供了三个构造方法。
int capacity： 表示数组的长度，也就是队列的长度。
boolean fair：表示是否为公平的阻塞队列，默认情况下构造的是非公平的阻塞队列。
Collection<? extends E> c：接收初始化数据。
public ArrayBlockingQueue(int capacity) {
...
}
public ArrayBlockingQueue(int capacity, boolean fair) {
...
}
public ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c) {
...
}
-
Add 方法
以 add 方法作为入口，在 add 方法中会调用父类的 add 方法，也就是 AbstractQueue。
一般这种写法都是调用父类的模版方法来解决通用性问题。
//ArrayBlockingQueue
public boolean add(E e) {
return super.add(e);
}
//AbstractQueue
public boolean add(E e) {
if (offer(e))
return true;
else
//如果队列满了直接抛出一个异常
throw new IllegalStateException("Queue full");
}
- offer 方法
- add 方法最终调用 offer 方法来添加数据，返回一个添加成功或者失败的布尔值反馈。
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();
}
}
- 判断添加的数据是否为空。
- 添加重入锁。
- 判断队列长度，如果队列长度等于数组长度，表示满了直接返回 false。
- 否则直接调用 enqueue 将元素添加到队列中。
- enqueue 方法
- 这个是最核心的逻辑，方法内部通过 putIndex 索引直接将元素添加到数组 items
private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;
items[putIndex] = x;
// 当putIndex 等于数组长度时，将 putIndex 重置为 0
if (++putIndex == items.length)
putIndex = 0;
count++;
notEmpty.signal();
}
- 由于 ArrayBlockingQueue 是一个 FIFO 的队列，队列添加元素时，是从队尾获取 putIndex 来存储元素，当 putIndex等于数组长度时，下次就需要从数组头部开始添加了。
- 当元素满了以后是无法继续添加的，因为会报错。
- 队列中的元素肯定会有一个消费者线程通过 take 或者其他方法来获取数据，而获取数据的同时元素也会从队列中移除。
- put 方法
- put 方法和 add 方法功能一样，差异是 put 方法如果队列满了，会阻塞。
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
//优先允许在等待时由其他线程调用等待线程的 interrupt 方法来中断等待直接返回。
lock.lockInterruptibly();
try {
while (count == items.length)
notFull.await();
enqueue(e);
} finally {
lock.unlock();
}
}
- take 方法
- 一种阻塞获取队列中元素的方法。
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
//可以中断。
lock.lockInterruptibly();
try {
//如果队列没有数据那么就加入 notEmpty 条件队列等待。
while (count == 0)
notEmpty.await();
//有数据就直接取走。
return dequeue();
} finally {
lock.unlock();
}
}
在等待过程中，如果队列中添加了元素，那么这个时候，会在 enqueue 中调用 notempty.signal 唤醒 take 线程来获得元素。
- dequeue 方法
- 这个是出队列的方法，主要是删除队列头部的元素并发返回给客户端。
private E dequeue() {
// assert lock.getHoldCount() == 1;
// assert items[takeIndex] != null;
final Object[] items = this.items;
@SuppressWarnings("unchecked")
//默认获取 0 位置的元素
E x = (E) items[takeIndex];
items[takeIndex] = null;
if (++takeIndex == items.length)
takeIndex = 0;
count--;
if (itrs != null)
//更新迭代器中的元素数据
itrs.elementDequeued();
//唤醒因为队列满了以后导致的被阻塞的线程
notFull.signal();
return x;
}
takeIndex，是用来记录拿数据的索引值
- 迭代器
- ArrayBlockingQueue 中，实现了迭代器的功能，也就是可以通过迭代器来遍历阻塞队列中的元素。
class Itrs {
...
void elementDequeued() {
// assert lock.getHoldCount() == 1;
if (count == 0)
queueIsEmpty();
else if (takeIndex == 0)
takeIndexWrapped();
}
}
itrs.elementDequeued() 是用来更新迭代器中的元素数据的。
- remove 方法
- remove 方法是移除一个指定元素。
public boolean remove(Object o) {
if (o == null) return false;
final Object[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count > 0) {
//获取下一个要添加元素时的索引。
final int putIndex = this.putIndex;
//获取当前要被移除的元素的索引。
int i = takeIndex;
do {
if (o.equals(items[i])) {
//移除指定元素。
removeAt(i);
return true;
}
if (++i == items.length)
i = 0;
//继续查找，直到找到最后一个元素。
} while (i != putIndex);
}
return false;
} finally {
lock.unlock();
}
}
原子操作类
原子性表示一个或者多个操作，要么全部执行完，要么一个也不执行。不能出现成功一部分失败一部分的情况。
多线程里面，要实现原子性，有几种方法，其中一种就是加 Synchronized 同步锁。
从 JDK1.5 开始，在 J.U.C 包中提供了 Atomic 包，提供了对于常用数据结构的原子操作。它提供了简单、高效、以及线程安全的更新一个变量的方式。
J.U.C 中的原子操作类
由于变量类型的关系，在 J.U.C 中提供了 12 个原子操作的类。这 12 个类可以分为四大类。
-
原子更新基本类型
- AtomicBoolean、AtomicInteger、AtomicLong
-
原子更新数组
- AtomicIntegerArray 、 AtomicLongArray 、AtomicReferenceArray
-
原子更新引用
- AtomicReference 、 AtomicReferenceFieldUpdater 、AtomicMarkableReference（更新带有标记位的引用类型）
-
原子更新字段
- AtomicIntegerFieldUpdater、AtomicLongFieldUpdater、AtomicStampedReference
AtomicInteger 原理分析
- getAndIncrement 方法
- getAndIncrement 实际上是调用 unsafe 这个类里面提供的方法。
public final int getAndIncrement() {
return unsafe.getAndAddInt(this, valueOffset, 1);
}
- getAndAddInt 方法
- 通过 do/while 循环，基于 CAS 乐观锁来做原子递增。
//Unsafe（反编译）
public final int getAndAddInt(Object var1, long var2, int var4) {
int var5;
do {
var5 = this.getIntVolatile(var1, var2);
} while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));
return var5;
}
- get 方法
- get 方法只需要直接返回 value 的值就行，这里的 value 是通过 Volatile 修饰的，用来保证可见性
public final int get() {
return value;
}
- 其他方法
- compareAndSet ， 允许客户端基于 AtomicInteger 来实现乐观锁的操作
public final boolean compareAndSet(int expect, int update) {
return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}
