详细分析Java并发集合LinkedBlockingQueue的用法

  /**
   * 链表的节点，同时也是通过它来实现一个单向链表
   */
  static class Node<E> {
    E item;

    // 指向链表的下一个节点
    Node<E> next;

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

   /**
   * 它的next指向队列头，这个节点不存储数据
   */
  transient Node<E> head;

  /**
   * 队列尾节点
   */
  private transient Node<E> last;

last = head = new Node<E>(null);

 /**
   * 向队列尾插入节点
   */
  private void enqueue(Node<E> node) {
    // assert putLock.isHeldByCurrentThread(); // 当前线程肯定获取了putLock锁
    // 将原队列尾节点的next引用指向新节点node，然后再将node节点设置成队列尾节点last
    // 这样就实现了向队列尾插入节点
    last = last.next = node;
  }

  // 移除队列头节点，并返回被删除的节点数据
  private E dequeue() {
    // assert takeLock.isHeldByCurrentThread(); // 当前线程肯定获取了takeLock锁
    // assert head.item == null;

    Node<E> h = head;
    // first节点中才存储了队列中第一个元素的数据
    Node<E> first = h.next;
    h.next = h; // help GC
    // 设置新的head值，相当于删除了first节点。因为head节点本身不储存数据
    head = first;
    // 队列头的数据
    E x = first.item;
    // 移除原先的数据
    first.item = null;
    return x;
  }

  /** 独占锁，用于处理插入队列操作的并发问题，即put与offer操作 */
  private final ReentrantLock putLock = new ReentrantLock();

  /** 队列不满的条件Condition，它是由putLock锁生成的 */
  private final Condition notFull = putLock.newCondition();

  /** 独占锁，用于处理删除队列头操作的并发问题，即take与poll操作 */
  private final ReentrantLock takeLock = new ReentrantLock();

  /** 队列不为空的条件Condition， 它使用takeLock锁生成的 */
  private final Condition notEmpty = takeLock.newCondition();

  /** 队列中元素的个数，这里使用AtomicInteger变量，保证并发安全问题 */
  private final AtomicInteger count = new AtomicInteger();

  /**
   * 唤醒在notEmpty条件下等待的线程，即移除队列头时，发现队列为空而被迫等待的线程。
   * 注意，因为要调用Condition的signal方法，必须获取对应的锁，所以这里调用了takeLock.lock()方法。
   * 当队列中插入元素(即put或offer操作)，那么队列肯定不为空，就会调用这个方法。
   */
  private void signalNotEmpty() {
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
      notEmpty.signal();
    } finally {
      takeLock.unlock();
    }
  }

  /**
   * 唤醒在notFull条件下等待的线程，即队列尾添加元素时，发现队列已满而被迫等待的线程。
   * 注意，因为要调用Condition的signal方法，必须获取对应的锁，所以这里调用了putLock.lock()方法
   * 当队列中删除元素(即take或poll操作)，那么队列肯定不满，就会调用这个方法。
   */
  private void signalNotFull() {
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
      notFull.signal();
    } finally {
      putLock.unlock();
    }
  }

  public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    // 记录插入操作前元素的个数
    int c = -1;
    // 创建新节点node
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    putLock.lockInterruptibly();
    try {
      //表示队列已满，那么就要调用notFull.await方法，让当前线程等待
      while (count.get() == capacity) {
        notFull.await();
      }
      // 向队列尾插入新元素
      enqueue(node);
      // 将当前队列元素个数加1，并返回加1之前的元素个数。
      c = count.getAndIncrement();
      // c + 1 < capacity表示队列未满，就唤醒可能等待插入操作的线程
      if (c + 1 < capacity)
        notFull.signal();
    } finally {
      putLock.unlock();
    }
    // c == 0表示插入之前，队列是空的。队列从空到放入一个元素时，
    // 才唤醒等待删除的线程
    // 防止频繁获取takeLock锁，消耗性能
    if (c == 0)
      signalNotEmpty();
  }

  public E take() throws InterruptedException {
    E x;
    int c = -1;
    final AtomicInteger count = this.count;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lockInterruptibly();
    try {
      //表示队列为空，那么就要调用notEmpty.await方法，让当前线程等待
      while (count.get() == 0) {
        notEmpty.await();
      }
      // 删除队列头元素，并返回它
      x = dequeue();
      // 返回当前队列个数，然后将队列个数减一
      c = count.getAndDecrement();
      // c > 1表示队列不为空，就唤醒可能等待删除操作的线程
      if (c > 1)
        notEmpty.signal();
    } finally {
      takeLock.unlock();
    }
    /**
     * c == capacity表示删除操作之前，队列是满的。只有从满队列中删除一个元素时，
     * 才唤醒等待插入的线程
     * 防止频繁获取putLock锁，消耗性能
     */
    if (c == capacity)
      signalNotFull();
    return x;
  }

  // 查看队列头元素
  public E peek() {
    // 队列为空，返回null
    if (count.get() == 0)
      return null;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
      // 获取队列头节点first
      Node<E> first = head.next;
      // first == null表示队列为空，返回null
      if (first == null)
        return null;
      else
        // 返回队列头元素
        return first.item;
    } finally {
      takeLock.unlock();
    }
  }

  // 从队列中删除指定元素o
  public boolean remove(Object o) {
    if (o == null) return false;
    // 因为不是删除列表头元素，所以就涉及到head和last两个变量，
    // putLock与takeLock都要加锁
    fullyLock();
    try {
      // 遍历整个队列，p表示当前节点，trail表示当前节点的前一个节点
      // 因为是单向链表，所以需要记录两个节点
      for (Node<E> trail = head, p = trail.next;
         p != null;
         trail = p, p = p.next) {
        // 如果找到了指定元素，那么删除节点p
        if (o.equals(p.item)) {
          unlink(p, trail);
          return true;
        }
      }
      return false;
    } finally {
      fullyUnlock();
    }
  }

  // 删除当前节点p，trail代表p的前一个节点
  void unlink(Node<E> p, Node<E> trail) {
    // 将当前节点的数据设置为null
    p.item = null;
    // 这样就在链表中删除了节点p
    trail.next = p.next;
    // 如果节点p是队列尾last，而它被删除了，那么就要将trail设置为last
    if (last == p)
      last = trail;
    // 将元素个数减一，如果原队列是满的，那么就调用notFull.signal()方法
    // 其实这个不用判断直接调用的，因为这里肯定获取了putLock锁
    if (count.getAndDecrement() == capacity)
      notFull.signal();
  }