玖叶教程网

前端编程开发入门

2024年04月21日

阻塞队列之LinkedBlockingQueue详解

1 简介

LinkedBlockingQueue是一个基于链表实现的阻塞队列，默认情况下，该阻塞队列的大小为Integer.MAX_VALUE，由于这个数值特别大，所以 LinkedBlockingQueue 也被称作无界队列，代表它几乎没有界限，队列可以随着元素的添加而动态增长，但是如果没有剩余内存，则队列将抛出OOM错误。所以为了避免队列过大造成机器负载或者内存爆满的情况出现，我们在使用的时候建议手动传一个队列的大小。

LinkedBlockingQueue内部由单链表实现，只能从head取元素，从tail添加元素。LinkedBlockingQueue采用两把锁的锁分离技术实现入队出队互不阻塞，添加元素和获取元素都有独立的锁，也就是说LinkedBlockingQueue是读写分离的，读写操作可以并行执行。

下面我通过测试用例以及对应的方法源码来讲解下。

2 测试用例

2.1 初始化

LinkedBlockingQueue<String> lbq = new LinkedBlockingQueue<>(1);

2.2 add方法

使用：

// 代码
@Test
public void add(){
  lbq.add("面试题解析");
  lbq.add("面试题解析");
}
// 输出
java.lang.IllegalStateException: Queue full
	at java.util.AbstractQueue.add(AbstractQueue.java:98)
	at LinkedBlockingQueueTest.add(LinkedBlockingQueueTest.java:14)
	...

源码解析：

// 1 调用AbstractQueue的add方法
public boolean add(E e) {
  // 当offer方法返回false的时候，就抛出了Queue full的异常。
  if (offer(e))
    return true;
  else
    throw new IllegalStateException("Queue full");
}
// 2 看下offer方法
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 {
    // 当第一次调用的时候，可以加入到队列中去，这个返回true
    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;
}
// 3 看下enqueue的方法，把node加入到队列中去
private void enqueue(Node<E> node) {
  // assert putLock.isHeldByCurrentThread();
  // assert last.next == null;
  last = last.next = node;
}

2.3 offer方法

使用：

// 代码
@Test
public void offer(){
  boolean offer1 = lbq.offer("面试题解析");
  log.info(offer1);
  boolean offer2 = lbq.offer("面试题解析");
  log.info(offer2);
}
// 输出
INFO - true
INFO - false

源码解析：

public boolean offer(E e) {
  if (e == null) throw new NullPointerException();
  final AtomicInteger count = this.count;
  // 当第二次加入时，元素个数和数组大小相同时，返回false，可以说明为啥offer2是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 {
    // 当第一次添加时，可以成功加入到队列，返回true，即offer1为true
    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;
}

2.4 put方法

使用：

// 代码
@Test
public void put() throws Exception{
  lbq.put("面试题解析");
  log.info("第一次put结束");
  lbq.put("面试题解析");
  log.info("第二次put结束");
}
// 输出
INFO - 第一次put结束

由结果可以看出程序阻塞在第二次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<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();
    }
    enqueue(node);
    c = count.getAndIncrement();
    if (c + 1 < capacity)
      notFull.signal();
  } finally {
    putLock.unlock();
  }
  if (c == 0)
    signalNotEmpty();
}

2.5 take方法

使用：

// 代码
@Test
public void take() throws Exception{
  log.info("开始take");
  String take = lbq.take();
  log.info(take);
}
// 输出
INFO - 开始take

从结果可以看出take没有输出，说明阻塞在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();
    if (c > 1)
      notEmpty.signal();
  } finally {
    takeLock.unlock();
  }
  if (c == capacity)
    signalNotFull();
  return x;
}

2.6 remove方法

使用：

// 代码
@Test
public void remove(){
  lbq.offer("面试题解析");
  String remove = lbq.remove();
  log.info(remove);
}
// 输出
INFO - 面试题解析

可以看出remove方法返回了被remove的值。

源码解析：

public E remove() {
  // 返回队列的首值，当队列为空时，则会报异常。
  E x = poll();
  if (x != null)
    return x;
  else
    throw new NoSuchElementException();
}

2.7 element方法

使用：

// 代码
@Test
public void element(){
  String element = lbq.element();
  log.info(element);
}
// 输出
java.util.NoSuchElementException
	at java.util.AbstractQueue.element(AbstractQueue.java:136)
	at LinkedBlockingQueueTest.element(LinkedBlockingQueueTest.java:49)
	...

说明当队列为空时，检测首值时会报错。

源码解析：

public E element() {
  // 返回队列的首值，当队列为空时，则会报异常。
  E x = peek();
  if (x != null)
    return x;
  else
    throw new NoSuchElementException();
}
// peek只会返回首值，而不会删除首值，这个是与poll的根本区别
public E peek() {
  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();
  }
}
// poll方法会使用dequeue方法返回值
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;
}
// dequeue方法会删除首值
private E dequeue() {
  // assert takeLock.isHeldByCurrentThread();
  // assert head.item == null;
  Node<E> h = head;
  Node<E> first = h.next;
  h.next = h; // help GC
  head = first;
  E x = first.item;
  first.item = null;
  return x;
}

2.8 remove方法

使用：

// 代码
@Test
public void peek(){
  String peek = lbq.peek();
  log.info(peek);
}
// 输出
INFO - null

队列为空时，返回的是null，讲解2.7 element时已经介绍了peek的源码，在此就不再赘述了。

2.9 remainingCapacity方法

使用：

// 代码
@Test
public void remainingCapacity(){
  int remainingCapacity = lbq.remainingCapacity();
  log.info(remainingCapacity);
}
// 输出
INFO - 1

求队列的剩余容量。

源码解析：

public int remainingCapacity() {
  // 返回的是数组的大小减去元素的个数
  return capacity - count.get();
}

2.10 clear方法

使用：

@Test
public void clear(){
  lbq.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();
  }
}

3 总结

LinkedBlockingQueue是一个可以指定容量的无界阻塞队列，使用锁分离技术，存取互不干扰，先进先出，入队由last指针记录，出队由head指针记录，而且线程池中也采用LinkedBlockingQueue阻塞队列，从而大大提高队列的吞吐量。

【温馨提示】

点赞+收藏文章，关注我并私信回复【面试题解析】，即可100%免费领取楼主的所有面试题资料！

标签：blockingqueue 

作者:gojiuye , 分类:技术精选 , 浏览:18 , 评论:0

发表评论:

控制面板
您好，欢迎到访网站！
  查看权限
网站分类
最新留言