多线程应用 - 阻塞队列LinkedBlockingQueue详解

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {

//内部类 Node结点
static class Node<E> {

    //当前结点中的元素
    E item;
    //下一个结点
    Node<E> next;
    Node(E x) {
        item = x;
    }
}

//队列总容量
private final int capacity;
//队列当前容量 使用了AtomicInteger来保证+1操作的原子性
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();

//无参构造函数 会指定队列最满容量为最大值
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);
}

//带指定集合的初始化构造函数
public LinkedBlockingQueue(Collection<? extends E> c) {
    //先调用无参构造函数 创建一个最满容量为最大值的队列
    this(Integer.MAX_VALUE);
    //获取写锁
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        int n = 0;
        //遍历元素集合
        for (E e : c) {
//这一段说明 和ArrayLinkedQueueLinkedBlockingDeque一样             ，元素不能为null
            if (e == null)
                throw new NullPointerException();
            //初始化时容量满了也会抛出异常
            if (n == capacity)
                throw new IllegalStateException("Queue full");
            //依次放入元素
            enqueue(new Node<E>(e));
            //计数+1
            ++n;
        }
        //构造函数中，如果保证单例或者线程封闭的情况，可以不需要考虑原子性，直接赋值
        count.set(n);
    } finally {
        //解锁
        putLock.unlock();
    }
}

//唤醒非空条件 让读操作获取锁 继续执行读操作
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<E> node) {
    last = last.next = node;
}

//查询并去除头结点元素
private E dequeue() {
    //获取头结点
    Node<E> h = head;
    //头结点的下一个结点
    Node<E> first = h.next;
    //头结点的下一个结点指向自己，方便gc
    h.next = h;
    //重新设置头结点为原来头结点的下一个结点
    head = first;
    //获取原来头结点元素
    E x = first.item;
    //将原头结点元素置为null
    first.item = null;
    //返回原头结点元素
    return x;
}

//读写锁都加锁，加锁顺序先写锁，再读锁
void fullyLock() {
    putLock.lock();
    takeLock.lock();
}

//读写锁都解锁，解锁顺序先读锁，再加锁
void fullyUnlock() {
    takeLock.unlock();
    putLock.unlock();
}//获取当前使用容量大小
public int size() {
    return count.get();
}

//获取剩余容量大小
public int remainingCapacity() {
    return capacity - count.get();
}

//存放元素
public void put(E e) throws InterruptedException {
    //元素不能为null
    if (e == null) throw new NullPointerException();
    int c = -1;
    //将当前元素构造成node结点
    Node<E> node = new Node<E>(e);
    //获取写锁
    final ReentrantLock putLock = this.putLock;
    //当前容量大小
    final AtomicInteger count = this.count;
    //获取响应中断锁
    putLock.lockInterruptibly();
    try {
        //容量已满
        while (count.get() == capacity) {
            //阻塞当前的写操作
            notFull.await();
        }
        //放入元素
        enqueue(node);
        //将容量+1，并返回操作前的容量
        c = count.getAndIncrement();
        //当前容量未满
        if (c + 1 < capacity)
            //唤醒阻塞的写操作线程
            notFull.signal();
    } finally {
        //解锁
        putLock.unlock();
    }
    if (c == 0)
        //原来队列容量若为0，这一步的写操作需要唤醒读操作
        signalNotEmpty();
}

//存放元素 与上述方法的区别：增加了返回值指定了阻塞时间，到达阻塞时间没被唤醒返回false
public boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException {
    //总体思想与上述put方法相同
    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) {
            //阻塞时间到期，返回false
            if (nanos <= 0)
                return false;
            //阻塞线程 并指定阻塞时间
            nanos = notFull.awaitNanos(nanos);
        }
        enqueue(new Node<E>(e));
        c = count.getAndIncrement();
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();
    }
    if (c == 0)
        //原来队列容量若为0，这一步的写操作需要唤醒读操作
        signalNotEmpty();
    return true;
}

//这个方法与put方法相比，不会阻塞线程，队列满则直接返回。
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<E> node = new Node<E>(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 {
        //当前队列元素为null 则阻塞
        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;
}

//取出元素 阻塞设置超时时间，原理与take方法相似
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 {
        //当前队列元素为null 则阻塞
        while (count.get() == 0) {
            //阻塞时间到期，返回null
            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;
}

//读操作，如果因为队列为空读取失败 则直接返回null，不阻塞
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;
}

//1、删除元素
public boolean remove(Object o) {
    if (o == null) return false;
    //获取写锁和读锁
    fullyLock();
    try {
        //从头结点遍历队列结点
        for (Node<E> 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();
    }
}

void unlink(Node<E> p, Node<E> trail) {
    p.item = null;
    //删除操作 trial的下一个结点指向原p结点的下一个结点
    trail.next = p.next;
    if (last == p)
        //如果原来p是最后一个结点，那么新的尾结点为trial
        last = trail;
    if (count.getAndDecrement() == capacity)
        //原容量满则唤醒写操作线程
        notFull.signal();
}

//2、校验是否包含，思想与删除一样，并且也需要获取读写锁
public boolean contains(Object o) {
    if (o == null) return false;
    fullyLock();
    try {
        for (Node<E> p = head.next; p != null; p = p.next)
            if (o.equals(p.item))
                return true;
        return false;
    } finally {
        fullyUnlock();
    }
}

//3、转换为数组，也需要获得读写锁
public Object[] toArray() {
    fullyLock();
    try {
        int size = count.get();
        Object[] a = new Object[size];
        int k = 0;
        for (Node<E> p = head.next; p != null; p = p.next)
            a[k++] = p.item;
        return a;
    } finally {
        fullyUnlock();
    }
}

//4、队列元素拼接成String 也需要获得读写锁
public String toString() {
    fullyLock();
    try {
        Node<E> p = head.next;
        if (p == null)
            return "[]";
        StringBuilder sb = new StringBuilder();
        sb.append('[');
        for (; ; ) {
            E e = p.item;
            sb.append(e == this ? "(this Collection)" : e);
            p = p.next;
            if (p == null)
                return sb.append(']').toString();
            sb.append(',').append(' ');
        }
    } finally {
        fullyUnlock();
    }
}

//5、清除队列，也需要获取读写锁
public void clear() {
    fullyLock();
    try {
        for (Node<E> p, h = head; (p = h.next) != null; h = p) {
            h.next = h;
            p.item = null;
        }
        head = last;
        // assert head.item == null && head.next == null;
        if (count.getAndSet(0) == capacity)
            notFull.signal();
    } finally {
        fullyUnlock();
    }
}