Java - LinkedBlockingQueue 源码阅读笔记

in Tech Java

简介

 LinkedBlockingQueue是由链表结构组成的有界单端阻塞队列,即对元素 FIFO(先进先出)进行排序。新元素插入队列尾部,检索操作获取队列头部元素。对数据进行增删改查时,使用两把ReentrantLock锁确保线程安全的同时,实现读写分离效果提高性能。使用Condition使线程在适当的时候休眠并释放锁,直至收到同Condition的信号才重新获取锁并继续执行。可作为生产-消费模型使用。

LinkedBlockingQueue 类

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
    private static final long serialVersionUID = -6903933977591709194L;

    /** 默认整型最大值 */
    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();

    /** 适用 takes 操作,取数据的时候,发现队列为空则等待 */
    private final Condition notEmpty = takeLock.newCondition();

    /** 写锁 */
    private final ReentrantLock putLock = new ReentrantLock();

    /** 适用 puts 操作,添加数据的时候,发现队列已满则等待 */
    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(); // Never contended, but necessary for visibility
        try {
            int n = 0;
            for (E e : c) {
                if (e == null)
                    throw new NullPointerException();
                if (n == capacity)
                    throw new IllegalStateException("Queue full");
                enqueue(new Node<E>(e));
                ++n;
            }
            count.set(n);
        } finally {
            putLock.unlock();
        }
    }

入队

put() 方法

    public void put(E e) throws InterruptedException {
        // 不允许空值
        if (e == null) throw new NullPointerException();
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            // 队列已满
            while (count.get() == capacity) {
                // await 方法会使线程休眠并释放锁
                notFull.await();
            }
            // 插入队尾
            enqueue(node);
            // 计数器增加1,返回旧值
            c = count.getAndIncrement();
            // 原有队列元素个数加1小于容量
            if (c + 1 < capacity)
                // 发送notFull信号,以唤醒休眠中同Condition的线程
                notFull.signal();
        } finally {
            // 最后必须解锁
            putLock.unlock();
        }
        // 原有队列元素个数为0
        if (c == 0)
            signalNotEmpty();
    }

enqueue() 方法

    private void enqueue(Node<E> node) {
        // assert putLock.isHeldByCurrentThread();
        // assert last.next == null;
        // 直接插入队尾即可
        last = last.next = node;
    }

signalNotEmpty() 方法

    private void signalNotEmpty() {
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            // 发送notEmpty信号,以唤醒休眠中同Condition的线程
            notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
    }

出队

take() 方法

    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) {
                // await 方法会使线程休眠并释放锁
                notEmpty.await();
            }
            // 从队首取数据
            x = dequeue();
            // 计数器减少1,返回旧值
            c = count.getAndDecrement();
            //  原有队列元素个数大于1,发送 notEmpty 信号
            if (c > 1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        // 原有队列元素个数等于容量
        if (c == capacity)
            signalNotFull();
        return x;
    }

dequeue() 方法

    private E dequeue() {
        // assert takeLock.isHeldByCurrentThread();
        // assert head.item == null;
        // 头结点是值为null的节点,第二个节点才是存储数据的节点
        // 将第二节点作为起始节点赋值给head,然后取出第一个节点的值,再赋值为null
        Node<E> h = head;
        Node<E> first = h.next;
        h.next = h; // help GC
        head = first;
        E x = first.item;
        first.item = null;
        return x;
    }

signalNotFull() 方法

    private void signalNotFull() {
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            // 发送notFull信号,以唤醒休眠中同Condition的线程
            notFull.signal();
        } finally {
            putLock.unlock();
        }
    }

remove() 方法

    public boolean remove(Object o) {
        if (o == null) return false;
        // 读写全锁
        fullyLock();
        try {
            // trail可以理解为p的上一个节点
            for (Node<E> trail = head, p = trail.next;p != null;trail = p, p = p.next) {
                if (o.equals(p.item)) {
                    // 将p节点的下一个节点与trail节点链接
                    unlink(p, trail);
                    return true;
                }
            }
            return false;
        } finally {
            fullyUnlock();
        }
    }

unlink() 方法

    void unlink(Node<E> p, Node<E> trail) {
        // assert isFullyLocked();
        // p.next is not changed, to allow iterators that are
        // traversing p to maintain their weak-consistency guarantee.
        p.item = null;
        trail.next = p.next;
        // 相等说明删除的是尾结点
        if (last == p)
            last = trail;
        // 原有队列元素个数等于容量
        if (count.getAndDecrement() == capacity)
            // 发送notFull信号,以唤醒休眠中同Condition的线程
            notFull.signal();
    }

fullyLock() 方法

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

fullyUnlock() 方法

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