菜鸟看源码之LinkedBlockingQueue

首先扯点别的：已经不记得上次是什么时候做仰卧起坐的了，现在重新开始锻炼起来，腹肌还是得保持的。但是刚做了两天就感觉小腹酸疼。做20个都得咬牙坚持。明天继续做。

今天分析一下LinkedBlockingQueue的源码。LinkedBlockingQueue 是基于链表的阻塞队列，按照先进先出的顺序来排列元素。默认长度可以达到Integer.MAX_VALUE 。也可以指定LinkedBlockingQueue最大长度。添加的元素不能为null。

LinkedBlockingQueue的继承结构

节点类

static class Node<E> {
    

        E item;

        Node<E> next;

        Node(E x) { item = x; }
    }

有一点：头节点（下文一律称作head）的item为null。尾节点（下文一律称作last）的next为null。

相关成员变量：

    //容量上限，默认是Integer.MAX_VALUE
    private final int capacity;

    //元素的个数
    private final AtomicInteger count = new AtomicInteger();

     //头节点，head.item==null
    transient Node<E> head;

    //尾节点，last.next==null
    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();

构造方法

创建一个容量是Integer.MAX_VALUE的LinkedBlockingQueue

 public LinkedBlockingQueue() {
        this(Integer.MAX_VALUE);
    }

创建一个容量是capacity的LinkedBlockingQueue

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

创建一个容量是Integer.MAX_VALUE的LinkedBlockingQueue，并把集合c中的所有元素加入LinkedBlockingQueue。

public LinkedBlockingQueue(Collection<? extends E> c) {
        //初始化容量为Integer.MAX_VALUE
        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();
        }
    }

把元素添加到队列的末尾

private void enqueue(Node<E> node) {
        // assert putLock.isHeldByCurrentThread();
        // assert last.next == null;
        last = last.next = node;//等价于 last.next = node;last =last.next;
    }

在添加集合c中的元素到LinkedBlockingQueue 中的时候，如果集合c中存在元素为null，则抛出空指针异常，如果集合c中的元素个数超过了LinkedBlockingQueue的容量，抛出一个队列已满的异常。

相关方法

boolean offer(E e)

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();//添加完元素后count要加1。
                if (c + 1 < capacity)//如果添加元素以后，队列没有满
                    notFull.signal();//唤醒notFull上等待的线程，
            }
        } finally {
            //释放锁
            putLock.unlock();
        }
        if (c == 0)//表示此时队列不为空，唤醒notEmpty上等待的线程。
            signalNotEmpty();
        return c >= 0;
    }

唤醒notEmpty上等待的线程，这个方法只会在put或offer方法中被调用

 private void signalNotEmpty() {
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
    }

boolean offer(E e, long timeout, TimeUnit unit)

 public boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException {

        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) {
                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)
            signalNotEmpty();
        return true;
    }

这个方法和boolean offer(E e) 类似，如果在规定的时间内无法添加元素，就返回false。如果在等待过程中被中断，那么就抛出中断异常。

void put(E e) ：如果队列已满的话，这个方法会阻塞，直到成功添加元素，或者当前线程被中断。

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;
        //使用lockInterruptibly方法，可以响应中断
        putLock.lockInterruptibly();
        try {
            //如果队列已满，就阻塞
            while (count.get() == capacity) {
                notFull.await();
            }
            //直到队列不满的时候，才添加元素
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)//如果队列未满，则唤醒notFull上等待的线程
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)//表示队列不为空，唤醒notEmpty上等待的线程
            signalNotEmpty();
    }

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) {
                notEmpty.await();
            }
            //直到队列不为空，才获取数据
            x = dequeue();
            c = count.getAndDecrement();
            if (c > 1)//队列不为空，唤醒notEmpty上等待的线程
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)//队列未满，唤醒notFull上等待的线程
            signalNotFull();
        return x;
    }

获取并删除元素，嗲用这个方法的时候，当前线程持有takeLock。

private E dequeue() {
        // assert takeLock.isHeldByCurrentThread();
        // assert head.item == 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;
    }

唤醒notFull上等待的线程，这个方法只会在take或poll方法中 被调用

 private void signalNotFull() {
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            notFull.signal();
        } finally {
            putLock.unlock();
        }
    }

poll()

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) {
                //如果队列不为空，调用dequeue获取并删除元素
                x = dequeue();
                c = count.getAndDecrement();
                if (c > 1)//获取元素以后，如果队列不为空，唤醒notEmpty上等待的线程
                    notEmpty.signal();
            }
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)//获取元素以后，如果队列未满，唤醒notFull上等待的线程
            signalNotFull();
        return x;
    }

poll(long timeout, TimeUnit unit)

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 {
            while (count.get() == 0) {
                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;
    }

这个方法和poll() 类似，如果在规定的时间内不能获取元素，返回null。可以响应中断，如果被中断，则抛出中断异常。

E peek() ：获取队列中第一个元素

public E peek() {
        if (count.get() == 0)
            return null;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            Node<E> first = head.next;
            if (first == null)
                return null;
            else
                return first.item;
        } finally {
            takeLock.unlock();
        }
    }

参考链接：
[1] : http://blog.csdn.net/u010887744/article/details/73010691