阻塞队列

有界队列

有固定大小的队列

ArrayBlockingQueue

数组实现的有界阻塞队列 按照先进先出的原则 对元素进行排序

//capacity = 队列长度      fair = 公平非公平
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();
    }

    public ArrayBlockingQueue(int capacity, boolean fair,
                              Collection<? extends E> c) {
        this(capacity, fair);
        final ReentrantLock lock = this.lock;
        lock.lock(); // Lock only for visibility, not mutual exclusion
        try {
            int i = 0;
            try {
                for (E e : c) {
                    checkNotNull(e);
                    items[i++] = e;
                }
            } catch (ArrayIndexOutOfBoundsException ex) {
                throw new IllegalArgumentException();
            }
            count = i;
            putIndex = (i == capacity) ? 0 : i;
        } finally {
            lock.unlock();
        }
    }

LinkedBlockingQueue

链表实现的有界队列 默认最大长度为Integer.MAX_VALUE 按照先进先出的原则对元素进行排序

无界队列

没有固定大小的队列

PriorityBlockingQueue

支持优先级的无界阻塞队列 默认情况下元素按照自然顺序升序排序 也可以通过自定义类实现compareTo方法来指定元素排序规则 或者在初始化的时候进行排序

DelayQueue(延时队列)

优先级队列实现的无界阻塞队列

SynchronousQueue

不存储元素的阻塞队列 每一个put操作必须等待一个take操作 否则不能继续添加元素

LinkedTransferQueue

由链表实现的无界阻塞队列 TransferQueue 相对其他阻塞队列 多了 tryTransfer和transfer方法

LinkedBlockingQueue

由链表实现的双向阻塞队列 双向阻塞队列的好处是多线程入队时 可以减少竞争

Put（）

队列满了 阻塞当前添加元素的线程

   public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        // Note: convention in all put/take/etc is to preset local var
        // holding count negative to indicate failure unless set.
        int c = -1;
       //根据添加的元素构建一个node
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        //加锁 
        putLock.lockInterruptibly();
        try {
            /*
             * Note that count is used in wait guard even though it is
             * not protected by lock. This works because count can
             * only decrease at this point (all other puts are shut
             * out by lock), and we (or some other waiting put) are
             * signalled if it ever changes from capacity. Similarly
             * for all other uses of count in other wait guards.
             */
            //如果队列元素满了 阻塞
            while (count.get() == capacity) {
                notFull.await();
            }
            //将node 添加到链表中
            enqueue(node);
            //元素数递增
            c = count.getAndIncrement();
            //当队列中元素小于最大容量的时候唤醒阻塞的生产者线程
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            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 {
            //如果队列长度== 0 阻塞
            while (count.get() == 0) {
                notEmpty.await();
            }
            //从队列中获取元素
            x = dequeue();
            //元素数递减
            c = count.getAndDecrement();
            //如果队列长度>1 唤醒处于阻塞状态下的消费者线程
            if (c > 1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }