ArrayBlockingQueue和LinkedBlockingQueue

ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(20); queue.put("element");//阻塞方法
        String element = queue.take();//阻塞方法

public void put(E e) throws InterruptedException { checkNotNull(e); final ReentrantLock lock = this.lock; lock.lockInterruptibly();//获取锁，如果interrupt，直接抛出异常
    try { while (count == items.length) notFull.await();//队列满了，阻塞等待被唤醒
        enqueue(e);//入队
    } finally { lock.unlock();//释放锁
 } } private void enqueue(E x) { final Object[] items = this.items; items[putIndex] = x; if (++putIndex == items.length)//如果入队下标已满，重置putIndex
            putIndex = 0; count++;//队列总数+1
        notEmpty.signal();//唤醒取队列阻塞的线程
}

public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly();//获取锁，如果interrupt，直接抛出异常
    try { while (count == 0) notEmpty.await();//队列空了，阻塞等待被唤醒
        return dequeue();//出队
    } finally { lock.unlock();//释放锁
 } } private E dequeue() { final Object[] items = this.items; @SuppressWarnings("unchecked") E x = (E) items[takeIndex]; items[takeIndex] = null; if (++takeIndex == items.length)//如果出队下标已满，重置takeIndex
            takeIndex = 0; count--;//总数减1 //迭代器有关
        if (itrs != null) itrs.elementDequeued(); notFull.signal();//唤醒入队列阻塞的线程
        return x; }

    public E poll() { final ReentrantLock lock = this.lock; lock.lock(); try { return (count == 0) ? null : dequeue(); } finally { lock.unlock(); } }

    //如果没有指定大小，默认为Integer.MAX_VALUE，也就是无界队列
    LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>(4);
    queue.put("element");//阻塞方法
    queue.take();//阻塞方法

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锁，如果interrupt，直接抛出异常
 putLock.lockInterruptibly(); try { while (count.get() == capacity) { notFull.await();//队列满了，阻塞等待被唤醒
 } enqueue(node);//入队
        c = count.getAndIncrement(); // 再次判断队列是否有可用空间，如果有唤醒下一个线程进行添加操作
        if (c + 1 < capacity) notFull.signal(); } finally { putLock.unlock();//释放putLock锁
 } if (c == 0)//唤醒取队列阻塞的线程
 signalNotEmpty(); } private void enqueue(Node<E> node) { //新增节点变为最后一个节点
        last = last.next = node; }

public E take() throws InterruptedException { E x; int c = -1; final AtomicInteger count = this.count;//总数 final ReentrantLock takeLock = this.takeLock; takeLock.lockInterruptibly(); //获取takeLock锁，如果interrupt，直接抛出异常
    try { while (count.get() == 0) { notEmpty.await();//队列空了，阻塞等待被唤醒
 } x = dequeue();//出队
        c = count.getAndDecrement(); //c > 1说明c至少为2， 队列中还有元素，唤醒下一个消费线程进行消费
        if (c > 1) notEmpty.signal(); } finally { takeLock.unlock();//释放takeLock锁
 } if (c == capacity)//唤醒入队列阻塞的线程
 signalNotFull(); return x; } private E dequeue() { // 获取到head节点
        Node<E> h = head; // 获取到head节点指向的下一个节点
        Node<E> first = h.next; // head节点原来指向的节点的next指向自己，等待下次gc回收
        h.next = h; // help GC // head节点指向新的节点
        head = first; // 获取到新的head节点的item值,即队列中的元素
        E x = first.item; first.item = null; return x; }