LinkedBlockingQueue详解

LinkedBlockingQueue简介

LinkedBlockingQueue类是堵塞/非堵塞（看你用啥方法），有边界线程安全队列。由下类图可以看到他的基本实现。它也是由单向链表实现，也是分别由两个Node指向head和tail，内部还有初始量为0的AtomicInteger的原子变量来计算队列数量。还有两个ReentrantLock独占锁来控制元素入队和出队的原子性。

源码详解

成员分析

//队列数量最大值默认为0x7FFFFFFF 对应 Integer.MAX_VALUE 即 2147483647
private final int capacity;

//原子性变量计算队列元素个数
private final AtomicInteger count = new AtomicInteger();
		
//头部节点
transient Node<E> head;

//尾部节点
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();

构造方法分析

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

public LinkedBlockingQueue(int capacity) {
    if (capacity <= 0) throw new IllegalArgumentException();
    this.capacity = capacity;
    //初始化首尾节点都指向哨兵节点
    last = head = new Node<E>(null);
}

入队操作分析

offer()

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();//自增
            if (c + 1 < capacity)
                notFull.signal();//唤醒因队列满而休眠的线程
        }
    } finally {
        putLock.unlock();//放锁
    }
    if (c == 0)
        signalNotEmpty();//唤醒因队列为空而休眠的线程
    return c >= 0;
}

offer()非堵塞入队方法，负责向队列尾部添加一个元素，如果队列有位置添加成功后返回true，否则返回false，如果元素为null则抛出NPE异常。

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 {
        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()方法向队列尾部添加一个元素，如果队列不为空且添加成功就返回true，否则堵塞线程，直至队列有位置添加成功在返回。如果在该线程被堵塞时设置了中断标志则会抛出InterruptedException异常，如果元素为null则抛出NPE异常。

出队操作分析

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){
            x = dequeue();出队
            c = count.getAndDecrement();递减
            if (c > 1)
                notEmpty.signal();//唤醒因队列为空而休眠的线程
        }
    } finally {
        takeLock.unlock();
    }
    if (c == capacity)
        signalNotFull();//唤醒因队列满而休眠的线程
    return x;
}

poll()方法从首部节点获取并移除一个元素，如果队列为空则返回null，否则返回元素值,另外该方法为非堵塞的。

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.signal();//唤醒因队列为空而休眠的线程
      } finally {
          takeLock.unlock();
      }
      if (c == capacity)
          signalNotFull();//唤醒因队列为满而休眠的线程
      return x;
  }

take()方法从首部节点获得并移除一个元素，如果队列为空则堵塞当前线程直到队列不为空返回元素值。如果当前线程被设置了中断标志则被堵塞线程会抛出InterruptedException异常,队列不为空则直接返回元素值。

其他操作

peek()

public E peek() {
    if (count.get() == 0)//队列为空提前返回null
        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();
    }
}

peek()方法从首部节点获得元素值，如果队列为空则返回null，否则返回元素值，另外该方法非堵塞。

size()

1
2
3

public int size() {
    return count.get();
}

size()获取当前队列元素个数

remove()

public boolean remove(Object o) {
    if (o == null) return false;//如果参数为null则直接返回false
    fullyLock();//这波是获得入队操作，出队操作的两把独占锁
    try {
    	  //遍历队列查找元素，发现则删除返回true，否则返回false
        for (Node<E> trail = head, p = trail.next;
             p != null;
             trail = p, p = p.next) {
            if (o.equals(p.item)) {
                unlink(p, trail);
                return true;
            }
        }
        return false;
    } finally {
        fullyUnlock();
    }
}