LinkedBlockingQueue 链表队列，其元素构成为：

static class Node<E> {

        E item;

        Node<E> next;

        Node(E x) { item = x; }

    }

该队列有两种锁及判断队列不为空和队列未满的条件对象

    /** 用于消费队列的锁，如操作：take,poll 等等 */

    private final ReentrantLock takeLock = new ReentrantLock();

    /** 用于消费时判断不为空的条件对象 */

    private final Condition notEmpty = takeLock.newCondition();

    /** 添加元素的锁，如操作：put，offer等等 */

    private final ReentrantLock putLock = new ReentrantLock();

    /** 用于添加元素时判断队列未满的条件对象 */

    private final Condition notFull = putLock.newCondition();

队列操作

添加元素

有三种方式：

1.offer

当队列已满时，添加失败，返回false，源码如下：

    public boolean offer(E e) {

        if (e == null) throw new NullPointerException();

        final AtomicInteger count = this.count;

        //判断队列是否已满，若队列已满，则返回false

        if (count.get() == capacity)

            return false;

        int c = -1;

        Node<E> node = new Node<E>(e);

        //获取放入元素的锁

        final ReentrantLock putLock = this.putLock;

        putLock.lock();

        try {

            //队列未满，这里是double检查，是获取锁后又检查了一次队列是否已满

            if (count.get() < capacity) {

                //添加元素到队尾

                enqueue(node);

                //获取添加元素前的队列大小，并将队列大小+1

                c = count.getAndIncrement();

                //判断队列未满

                if (c + 1 < capacity)

                    //则通知下一个线程继续添加元素

                    notFull.signal();

            }

        } finally {

            putLock.unlock();

        }

        //队列中有元素，则通知消费线程可以消费

        if (c == 0)

            signalNotEmpty();

        return c >= 0;

    }

2.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 {

            //这里是double检查，是获取锁后又检查了一次队列是否已满，如果队列已满，则线程进入阻塞，直到队列中有元素出队

            while (count.get() == capacity) {

                notFull.await();

            }

            //添加元素到队尾

            enqueue(node);

            c = count.getAndIncrement();

            //判断队列未满

            if (c + 1 < capacity)

                //通知其它线程添加元素

                notFull.signal();

        } finally {

            putLock.unlock();

        }

        //判断队列已有元素，则通知阻塞的消费线程进行消费

        if (c == 0)

            signalNotEmpty();

    }

put指定超时的操作

指定put的等待超时时间，等待超时后，则返回操作false

    //指定超时时间

    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) {

                //当超时时间过了后，则不再继续待，返回操作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)

            signalNotEmpty();

        return true;

    }

3.add

实际上执行的是offer操作，判断offer操作是否成功，若失败，则抛出队列已满的异常信息，其实现在父类AbstractQueue中

  public boolean add(E e) {

        if (offer(e))

            return true;

        else

            throw new IllegalStateException("Queue full");

    }

出队操作

1.take

与put相反，若队列为空，则阻塞等待，直到队列有元素入队

 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.signal();

        } finally {

            takeLock.unlock();

        }

        //判断队列未满，通知入队的线程进行入队操作

        if (c == capacity)

            signalNotFull();

        return x;

    }

take指定超时时间

当指定了take操作的超时时间后，take等待超时时，若队列还未有元素，则返回null

    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) {

                //超时时间已过，则返回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;

    }

2.poll

若队列为空，则直接返回null，否则返回队头元素

    public E poll() {

        final AtomicInteger count = this.count;

        //队列为空，则返回null

        if (count.get() == 0)

            return null;

        E x = null;

        int c = -1;

        final ReentrantLock takeLock = this.takeLock;

        //获取消费锁

        takeLock.lock();

        try {

            //double检查，再次判断队列不为空，则返回队头元素

            if (count.get() > 0) {

                x = dequeue();

                c = count.getAndDecrement();

                //队列还有元素，则通知其它消费线程进行消费操作

                if (c > 1)

                    notEmpty.signal();

            }

        } finally {

            takeLock.unlock();

        }

        //判断出队后，队列未满，则通知入队线程进行入队操作

        if (c == capacity)

            signalNotFull();

        return x;

    }

3.remove

poll操作，若返回null，则返回空队列异常，其操作在父类AbstractQueue中

    public E remove() {

        E x = poll();

        if (x != null)

            return x;

        else

            throw new NoSuchElementException();

    }

4.peek

返回队列第一个元素，元素并不出队。

    public E peek() {

        //队列为空，则返回null

        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.remove(object)

移除指定元素

    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);
                    //若队列中有指定元素，返回true

                    return true;

                }

            }
            //未找到指定元素，返回false

            return false;

        } finally {

            fullyUnlock();

        }

    }

2.contains(object)

判断队列是否存在指定元素

    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.toArray() ，toArray(T[] a)

将队列转换成数组

    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();

        }

    }

    public <T> T[] toArray(T[] a) {

        fullyLock();

        try {

            int size = count.get();

            if (a.length < size)

                a = (T[])java.lang.reflect.Array.newInstance

                    (a.getClass().getComponentType(), size);

            int k = 0;

            for (Node<E> p = head.next; p != null; p = p.next)

                a[k++] = (T)p.item;

            if (a.length > k)

                a[k] = null;

            return a;

        } finally {

            fullyUnlock();

        }

    }

4.clear()

清空队列

    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();

        }

    }

入队实现

    private void enqueue(Node<E> node) {
        //将原队尾元素指向新元素node，新元素的node的next为null

        last = last.next = node;

    }

出队实现

    private E dequeue() {
        //取队头元素

        Node<E> h = head;
        //取出队头元素的下一个元素

        Node<E> first = h.next;
        //将原队头元素的下一个元素设置指向为原队头，即本身node，即该node只有自已引用自己

        h.next = h; // help GC
        //设置新队头元素

        head = first;
        //取出节点数据

        E x = first.item;
        //将节点的数据设置为null，则该node除了自己引用自己外，无其它引用，可以被垃圾回收

        first.item = null;

        return x;

    }

巴特西

LinkedBlockingQueue 学习