Java 阻塞队列

不满足就阻塞等待

ArrayBlockingQueue

---

基于数组

ArrayBlockingQueue.java

public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {
    //底层是对象数组
    final Object[] items;

    //数组循环使用，因此要记录插入位置和取出位置
    int takeIndex;
    int putIndex;

    //记录个数
    int count;

    //主锁，锁定所有访问，插入和取出不能同时进行
    final ReentrantLock lock;
    //锁条件，用于控制插入取出时的阻塞行为
    private final Condition notEmpty;
    private final Condition notFull;

    //...
}

构造

ArrayBlockingQueue.java

//默认是非公平锁模式
public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
}
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();
}

插入数据

ArrayBlockingQueue.java

//入队核心逻辑
private void enqueue(E x) {
    final Object[] items = this.items;
    items[putIndex] = x;
    //下标归0循环使用
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
    //有数据，通知条件非空
    notEmpty.signal();
}

//非阻塞插入
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 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();
    }
}

取出数据

ArrayBlockingQueue.java

//出队核心逻辑
private E dequeue() {
    final Object[] items = this.items;
    //取出后清空数组位置
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    //下标归0循环使用
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    //有空间，通知条件非满
    notFull.signal();
    return x;
}

//非阻塞取出
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();
    }
}

LinkedBlockingQueue 和 LinkedBlockingDeque

---

单链表采取双锁结构，每个锁关联一个条件，这样插入和取出可以不冲突，吞吐量增强，代价是插入的数据不一定能及时被取出数据线程可见

LinkedBlockingQueue.java

public class LinkedBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {
    private final int capacity;
    //使用原子计数类，避免加锁
    private final AtomicInteger count = new AtomicInteger();

    transient Node<E> head;
    private transient Node<E> 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();
}

提供完全加解锁方法，供一些必须保证可见性的方法使用，比如remove

LinkedBlockingQueue.java

void fullyLock() {
    putLock.lock();
    takeLock.lock();
}
void fullyUnlock() {
    takeLock.unlock();
    putLock.unlock();
}

基于链表实现的队列实际也是有界的，不强制指定容量，默认是整型最大值

LinkedBlockingQueue.java

public LinkedBlockingQueue() {
    this(Integer.MAX_VALUE);
}
public LinkedBlockingQueue(int capacity) {
    if (capacity <= 0) throw new IllegalArgumentException();
    this.capacity = capacity;
    last = head = new Node<E>(null);
}

双向链表实现版本和数组类似，采用一锁加两条件方式

LinkedBlockingDeque.java

public class LinkedBlockingDeque<E> extends AbstractQueue<E> implements BlockingDeque<E>, java.io.Serializable {
    //链表相关
    transient Node<E> first;
    transient Node<E> last;

    //容量相关
    private transient int count;
    private final int capacity;

    //锁相关
    final ReentrantLock lock = new ReentrantLock();
    private final Condition notEmpty = lock.newCondition();
    private final Condition notFull = lock.newCondition();
    //...
}

一分也是爱~

---

版权声明

This site by Linest is licensed under a Creative Commons BY-NC-ND 4.0 International License.
由Linest创作并维护的博客采用创作共用保留署名-非商业-禁止演绎4.0国际许可证。

本文永久链接：http://linest.github.io/2018/01/21/java-multithread-blockingqueue/