经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » Java相关 » Java » 查看文章
多线程系列(十三) -一文带你搞懂阻塞队列
来源：cnblogs　　作者：程序员志哥　　时间：2024/3/7 9:14:53　　对本文有异议

一、摘要

在之前的文章中，我们介绍了生产者和消费者模型的最基本实现思路，相信大家对它已经有一个初步的认识。

在 Java 的并发包里面还有一个非常重要的接口：BlockingQueue。

BlockingQueue是一个阻塞队列，更为准确的解释是：BlockingQueue是一个基于阻塞机制实现的线程安全的队列。通过它也可以实现生产者和消费者模型，并且效率更高、安全可靠，相比之前介绍的生产者和消费者模型，它可以同时实现生产者和消费者并行运行。

那什么是阻塞队列呢？

简单的说，就是当参数在入队和出队时，通过加锁的方式来避免线程并发操作时导致的数据异常问题。

在 Java 中，能对线程并发执行进行加锁的方式主要有synchronized和ReentrantLock，其中BlockingQueue采用的是ReentrantLock方式实现。

与此对应的还有非阻塞机制的队列，主要是采用 CAS 方式来控制并发操作，例如：ConcurrentLinkedQueue，这个我们在后面的文章再进行分享介绍。

今天我们主要介绍BlockingQueue相关的知识和用法，废话不多说了，进入正题！

二、BlockingQueue 方法介绍

打开BlockingQueue的源码，你会发现它继承自Queue，正如上文提到的，它本质是一个队列接口。

  1. public interface BlockingQueue<E> extends Queue<E> {
  2. 	//...省略
  3. }

关于队列，我们在之前的集合系列文章中对此有过深入的介绍，本篇就再次简单的介绍一下。

队列其实是一个数据结构，元素遵循先进先出的原则，所有新元素的插入，也被称为入队操作，会插入到队列的尾部；元素的移除，也被称为出队操作，会从队列的头部开始移除，从而保证先进先出的原则。

在Queue接口中，总共有 6 个方法，可以分为 3 类，分别是：插入、移除、查询，内容如下：

方法 描述
add(e) 插入元素，如果插入失败，就抛异常
offer(e) 插入元素，如果插入成功，就返回 true；反之 false
remove() 移除元素，如果移除失败，就抛异常
poll() 移除元素，如果移除成功，返回 true；反之 false
element() 获取队首元素，如果获取结果为空，就抛异常
peek() 获取队首元素，如果获取结果为空，返回空对象

因为BlockingQueue是Queue的子接口，了解Queue接口里面的方法，有助于我们对BlockingQueue的理解。

除此之外，BlockingQueue还单独扩展了一些特有的方法，内容如下：

方法 描述
put(e) 插入元素，如果没有插入成功，线程会一直阻塞，直到队列中有空间再继续
offer(e, time, unit) 插入元素，如果在指定的时间内没有插入成功，就返回 false；反之 true
take() 移除元素，如果没有移除成功，线程会一直阻塞，直到队列中新的数据被加入
poll(time, unit) 移除元素，如果在指定的时间内没有移除成功，就返回 false；反之 true
drainTo(Collection c, int maxElements) 一次性取走队列中的数据到 c 中，可以指定取的个数。该方法可以提升获取数据效率，不需要多次分批加锁或释放锁

分析源码，你会发现相比普通的Queue子类，BlockingQueue子类主要有以下几个明显的不同点：

  • 1.元素插入和移除时线程安全：主要是通过在入队和出队时进行加锁，保证了队列线程安全，加锁逻辑采用ReentrantLock实现
  • 2.支持阻塞的入队和出队方法：当队列满时，会阻塞入队的线程，直到队列不满；当队列为空时，会阻塞出队的线程，直到队列中有元素；同时支持超时机制，防止线程一直阻塞

三、BlockingQueue 用法详解

打开源码，BlockingQueue接口的实现类非常多，我们重点讲解一下其中的 5 个非常重要的实现类，分别如下表所示。

实现类 功能
ArrayBlockingQueue 基于数组的阻塞队列，使用数组存储数据，需要指定长度，所以是一个有界队列
LinkedBlockingQueue 基于链表的阻塞队列，使用链表存储数据，默认是一个无界队列；也可以通过构造方法中的capacity设置最大元素数量，所以也可以作为有界队列
SynchronousQueue 一种没有缓冲的队列，生产者产生的数据直接会被消费者获取并且立刻消费
PriorityBlockingQueue 基于优先级别的阻塞队列，底层基于数组实现，是一个无界队列
DelayQueue 延迟队列，其中的元素只有到了其指定的延迟时间，才能够从队列中出队

下面我们对以上实现类的用法，进行一一介绍。

3.1、ArrayBlockingQueue

ArrayBlockingQueue是一个基于数组的阻塞队列，初始化的时候必须指定队列大小，源码实现比较简单，采用的是ReentrantLock和Condition实现生产者和消费者模型，部分核心源码如下：

  1. public class ArrayBlockingQueue<E> extends AbstractQueue<E>
  2.         implements BlockingQueue<E>, java.io.Serializable {
  4. 	/** 使用数组存储队列中的元素 */
  5. 	final Object[] items;
  7. 	/** 使用独占锁ReetrantLock */
  8. 	final ReentrantLock lock;
  10. 	/** 等待出队的条件 */
  11. 	private final Condition notEmpty;
  13. 	/** 等待入队的条件 */
  14. 	private final Condition notFull;
  16. 	/** 初始化时，需要指定队列大小 */
  17. 	public ArrayBlockingQueue(int capacity) {
  18.         this(capacity, false);
  19.     }
  21.     /** 初始化时，也指出指定是否为公平锁， */
  22.     public ArrayBlockingQueue(int capacity, boolean fair) {
  23.         if (capacity <= 0)
  24.             throw new IllegalArgumentException();
  25.         this.items = new Object[capacity];
  26.         lock = new ReentrantLock(fair);
  27.         notEmpty = lock.newCondition();
  28.         notFull =  lock.newCondition();
  29.     }
  31.     /**入队操作*/
  32.     public void put(E e) throws InterruptedException {
  33.         checkNotNull(e);
  34.         final ReentrantLock lock = this.lock;
  35.         lock.lockInterruptibly();
  36.         try {
  37.             while (count == items.length)
  38.                 notFull.await();
  39.             enqueue(e);
  40.         } finally {
  41.             lock.unlock();
  42.         }
  43.     }
  45.     /**出队操作*/
  46.     public E take() throws InterruptedException {
  47.         final ReentrantLock lock = this.lock;
  48.         lock.lockInterruptibly();
  49.         try {
  50.             while (count == 0)
  51.                 notEmpty.await();
  52.             return dequeue();
  53.         } finally {
  54.             lock.unlock();
  55.         }
  56.     }
  57. }

ArrayBlockingQueue采用ReentrantLock进行加锁，只有一个ReentrantLock对象，这意味着生产者和消费者无法并行运行。

我们看一个简单的示例代码如下：

  1. public class Container {
  3.     /**
  4.      * 初始化阻塞队列
  5.      */
  6.     private final BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);
  8.     /**
  9.      * 添加数据到阻塞队列
  10.      * @param value
  11.      */
  12.     public void add(Integer value) {
  13.         try {
  14.             queue.put(value);
  15.             System.out.println("生产者："+ Thread.currentThread().getName()+"，add：" + value);
  16.         } catch (InterruptedException e) {
  17.             e.printStackTrace();
  18.         }
  19.     }
  21.     /**
  22.      * 从阻塞队列获取数据
  23.      */
  24.     public void get() {
  25.         try {
  26.             Integer value = queue.take();
  27.             System.out.println("消费者："+ Thread.currentThread().getName()+"，value：" + value);
  28.         } catch (InterruptedException e) {
  29.             e.printStackTrace();
  30.         }
  31.     }
  33. }
  1. /**
  2.  * 生产者
  3.  */
  4. public class Producer extends Thread {
  6.     private Container container;
  8.     public Producer(Container container) {
  9.         this.container = container;
  10.     }
  12.     @Override
  13.     public void run() {
  14.         for (int i = 0; i < 6; i++) {
  15.             container.add(i);
  16.         }
  17.     }
  18. }
  1. /**
  2.  * 消费者
  3.  */
  4. public class Consumer extends Thread {
  6.     private Container container;
  8.     public Consumer(Container container) {
  9.         this.container = container;
  10.     }
  12.     @Override
  13.     public void run() {
  14.         for (int i = 0; i < 6; i++) {
  15.             container.get();
  16.         }
  17.     }
  18. }
  1. /**
  2.  * 测试类
  3.  */
  4. public class MyThreadTest {
  6.     public static void main(String[] args) {
  7.         Container container = new Container();
  9.         Producer producer = new Producer(container);
  10.         Consumer consumer = new Consumer(container);
  12.         producer.start();
  13.         consumer.start();
  14.     }
  15. }

运行结果如下：

  1. 生产者：Thread-0，add：0
  2. 生产者：Thread-0，add：1
  3. 生产者：Thread-0，add：2
  4. 生产者：Thread-0，add：3
  5. 生产者：Thread-0，add：4
  6. 生产者：Thread-0，add：5
  7. 消费者：Thread-1，value：0
  8. 消费者：Thread-1，value：1
  9. 消费者：Thread-1，value：2
  10. 消费者：Thread-1，value：3
  11. 消费者：Thread-1，value：4
  12. 消费者：Thread-1，value：5

可以很清晰的看到，生产者线程执行完毕之后，消费者线程才开始消费。

3.2、LinkedBlockingQueue

LinkedBlockingQueue是一个基于链表的阻塞队列，初始化的时候无须指定队列大小，默认队列长度为Integer.MAX_VALUE，也就是 int 型最大值。

同样的，采用的是ReentrantLock和Condition实现生产者和消费者模型，不同的是它使用了两个lock，这意味着生产者和消费者可以并行运行，程序执行效率进一步得到提升。

部分核心源码如下：

  1. public class LinkedBlockingQueue<E> extends AbstractQueue<E>
  2.         implements BlockingQueue<E>, java.io.Serializable {
  3.     /** 使用出队独占锁ReetrantLock */
  4.     private final ReentrantLock takeLock = new ReentrantLock();
  6.     /** 等待出队的条件 */
  7.     private final Condition notEmpty = takeLock.newCondition();
  9.     /** 使用入队独占锁ReetrantLock */
  10.     private final ReentrantLock putLock = new ReentrantLock();
  12.     /** 等待入队的条件 */
  13.     private final Condition notFull = putLock.newCondition();
  15.     /**入队操作*/
  16.     public void put(E e) throws InterruptedException {
  17.         if (e == null) throw new NullPointerException();
  18.         int c = -1;
  19.         Node<E> node = new Node<E>(e);
  20.         final ReentrantLock putLock = this.putLock;
  21.         final AtomicInteger count = this.count;
  22.         putLock.lockInterruptibly();
  23.         try {
  24.             while (count.get() == capacity) {
  25.                 notFull.await();
  26.             }
  27.             enqueue(node);
  28.             c = count.getAndIncrement();
  29.             if (c + 1 < capacity)
  30.                 notFull.signal();
  31.         } finally {
  32.             putLock.unlock();
  33.         }
  34.         if (c == 0)
  35.             signalNotEmpty();
  36.     }
  38.     /**出队操作*/
  39.     public E take() throws InterruptedException {
  40.         E x;
  41.         int c = -1;
  42.         final AtomicInteger count = this.count;
  43.         final ReentrantLock takeLock = this.takeLock;
  44.         takeLock.lockInterruptibly();
  45.         try {
  46.             while (count.get() == 0) {
  47.                 notEmpty.await();
  48.             }
  49.             x = dequeue();
  50.             c = count.getAndDecrement();
  51.             if (c > 1)
  52.                 notEmpty.signal();
  53.         } finally {
  54.             takeLock.unlock();
  55.         }
  56.         if (c == capacity)
  57.             signalNotFull();
  58.         return x;
  59.     }
  60. }

把最上面的样例Container中的阻塞队列实现类换成LinkedBlockingQueue，调整如下：

  1. /**
  2.  * 初始化阻塞队列
  3.  */
  4. private final BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();

再次运行结果如下：

  1. 生产者：Thread-0，add：0
  2. 消费者：Thread-1，value：0
  3. 生产者：Thread-0，add：1
  4. 消费者：Thread-1，value：1
  5. 生产者：Thread-0，add：2
  6. 消费者：Thread-1，value：2
  7. 生产者：Thread-0，add：3
  8. 生产者：Thread-0，add：4
  9. 生产者：Thread-0，add：5
  10. 消费者：Thread-1，value：3
  11. 消费者：Thread-1，value：4
  12. 消费者：Thread-1，value：5

可以很清晰的看到，生产者线程和消费者线程，交替并行执行。

3.3、SynchronousQueue

SynchronousQueue是一个没有缓冲的队列，生产者产生的数据直接会被消费者获取并且立刻消费，相当于传统的一个请求对应一个应答模式。

相比ArrayBlockingQueue和LinkedBlockingQueue，SynchronousQueue实现机制也不同，它主要采用队列和栈来实现数据的传递，中间不存储任何数据，生产的数据必须得消费者处理，线程阻塞方式采用 JDK 提供的LockSupport park/unpark函数来完成，也支持公平和非公平两种模式。

  • 当采用公平模式时：使用一个 FIFO 队列来管理多余的生产者和消费者
  • 当采用非公平模式时：使用一个 LIFO 栈来管理多余的生产者和消费者，这也是SynchronousQueue默认的模式

部分核心源码如下：

  1. public class SynchronousQueue<E> extends AbstractQueue<E>
  2.     implements BlockingQueue<E>, java.io.Serializable {
  4.     /**不同的策略实现*/
  5.     private transient volatile Transferer<E> transferer;
  7. 	/**默认非公平模式*/
  8.     public SynchronousQueue() {
  9.         this(false);
  10.     }
  12.     /**可以选策略，也可以采用公平模式*/
  13.     public SynchronousQueue(boolean fair) {
  14.         transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
  15.     }
  17. 	/**入队操作*/
  18.     public void put(E e) throws InterruptedException {
  19.         if (e == null) throw new NullPointerException();
  20.         if (transferer.transfer(e, false, 0) == null) {
  21.             Thread.interrupted();
  22.             throw new InterruptedException();
  23.         }
  24.     }
  26.     /**出队操作*/
  27.     public E take() throws InterruptedException {
  28.         E e = transferer.transfer(null, false, 0);
  29.         if (e != null)
  30.             return e;
  31.         Thread.interrupted();
  32.         throw new InterruptedException();
  33.     }
  34. }

同样的，把最上面的样例Container中的阻塞队列实现类换成SynchronousQueue，代码如下：

  1. public class Container {
  3.     /**
  4.      * 初始化阻塞队列
  5.      */
  6.     private final BlockingQueue<Integer> queue = new SynchronousQueue<>();
  9.     /**
  10.      * 添加数据到阻塞队列
  11.      * @param value
  12.      */
  13.     public void add(Integer value) {
  14.         try {
  15.             queue.put(value);
  16.             Thread.sleep(100);
  17.             System.out.println("生产者："+ Thread.currentThread().getName()+"，add：" + value);
  18.         } catch (InterruptedException e) {
  19.             e.printStackTrace();
  20.         }
  21.     }
  24.     /**
  25.      * 从阻塞队列获取数据
  26.      */
  27.     public void get() {
  28.         try {
  29.             Integer value = queue.take();
  30.             Thread.sleep(200);
  31.             System.out.println("消费者："+ Thread.currentThread().getName()+"，value：" + value);
  32.         } catch (InterruptedException e) {
  33.             e.printStackTrace();
  34.         }
  35.     }
  36. }

再次运行结果如下：

  1. 生产者：Thread-0，add：0
  2. 消费者：Thread-1，value：0
  3. 生产者：Thread-0，add：1
  4. 消费者：Thread-1，value：1
  5. 生产者：Thread-0，add：2
  6. 消费者：Thread-1，value：2
  7. 生产者：Thread-0，add：3
  8. 消费者：Thread-1，value：3
  9. 生产者：Thread-0，add：4
  10. 消费者：Thread-1，value：4
  11. 生产者：Thread-0，add：5
  12. 消费者：Thread-1，value：5

可以很清晰的看到，生产者线程和消费者线程，交替串行执行，生产者每投递一条数据，消费者处理一条数据。

3.4、PriorityBlockingQueue

PriorityBlockingQueue是一个基于优先级别的阻塞队列，底层基于数组实现，可以认为是一个无界队列。

PriorityBlockingQueue与ArrayBlockingQueue的实现逻辑，基本相似，也是采用ReentrantLock来实现加锁的操作。

最大不同点在于：

  • 1.PriorityBlockingQueue内部基于数组实现的最小二叉堆算法，可以对队列中的元素进行排序，插入队列的元素需要实现Comparator或者Comparable接口，以便对元素进行排序
  • 2.其次，队列的长度是可扩展的，不需要显式指定长度，上限为Integer.MAX_VALUE - 8

部分核心源码如下：

  1. public class PriorityBlockingQueue<E> extends AbstractQueue<E>
  2.     implements BlockingQueue<E>, java.io.Serializable {
  4.  	/**队列元素*/
  5.     private transient Object[] queue;
  7.     /**比较器*/
  8.     private transient Comparator<? super E> comparator;
  10.     /**采用ReentrantLock进行加锁*/
  11.     private final ReentrantLock lock;
  13.     /**条件等待与通知*/
  14.     private final Condition notEmpty;
  16.     /**入队操作*/
  17.     public boolean offer(E e) {
  18.         if (e == null)
  19.             throw new NullPointerException();
  20.         final ReentrantLock lock = this.lock;
  21.         lock.lock();
  22.         int n, cap;
  23.         Object[] array;
  24.         while ((n = size) >= (cap = (array = queue).length))
  25.             tryGrow(array, cap);
  26.         try {
  27.             Comparator<? super E> cmp = comparator;
  28.             if (cmp == null)
  29.                 siftUpComparable(n, e, array);
  30.             else
  31.                 siftUpUsingComparator(n, e, array, cmp);
  32.             size = n + 1;
  33.             notEmpty.signal();
  34.         } finally {
  35.             lock.unlock();
  36.         }
  37.         return true;
  38.     }
  40.     /**出队操作*/
  41.     public E take() throws InterruptedException {
  42.         final ReentrantLock lock = this.lock;
  43.         lock.lockInterruptibly();
  44.         E result;
  45.         try {
  46.             while ( (result = dequeue()) == null)
  47.                 notEmpty.await();
  48.         } finally {
  49.             lock.unlock();
  50.         }
  51.         return result;
  52.     }
  53. }

同样的，把最上面的样例Container中的阻塞队列实现类换成PriorityBlockingQueue，调整如下：

  1. /**
  2.  * 初始化阻塞队列
  3.  */
  4. private final BlockingQueue<Integer> queue = new PriorityBlockingQueue<>();

生产者插入数据的内容，我们改下插入顺序。

  1. /**
  2.  * 生产者
  3.  */
  4. public class Producer extends Thread {
  6.     private Container container;
  8.     public Producer(Container container) {
  9.         this.container = container;
  10.     }
  12.     @Override
  13.     public void run() {
  14.         container.add(5);
  15.         container.add(3);
  16.         container.add(1);
  17.         container.add(2);
  18.         container.add(0);
  19.         container.add(4);
  20.     }
  21. }

最后运行结果如下：

  1. 生产者：Thread-0，add：5
  2. 生产者：Thread-0，add：3
  3. 生产者：Thread-0，add：1
  4. 生产者：Thread-0，add：2
  5. 生产者：Thread-0，add：0
  6. 生产者：Thread-0，add：4
  7. 消费者：Thread-1，value：0
  8. 消费者：Thread-1，value：1
  9. 消费者：Thread-1，value：2
  10. 消费者：Thread-1，value：3
  11. 消费者：Thread-1，value：4
  12. 消费者：Thread-1，value：5

从日志上可以很明显看出，对于整数，默认情况下，按照升序排序，消费者默认从 0 开始处理。

3.5、DelayQueue

DelayQueue是一个线程安全的延迟队列，存入队列的元素不会立刻被消费，只有到了其指定的延迟时间，才能够从队列中出队。

底层采用的是PriorityQueue来存储元素，DelayQueue的特点在于：插入队列中的数据可以按照自定义的delay时间进行排序，快到期的元素会排列在前面，只有delay时间小于 0 的元素才能够被取出。

部分核心源码如下：

  1. public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
  2.     implements BlockingQueue<E> {
  4.     /**采用ReentrantLock进行加锁*/
  5.     private final transient ReentrantLock lock = new ReentrantLock();
  7.     /**采用PriorityQueue进行存储数据*/
  8.     private final PriorityQueue<E> q = new PriorityQueue<E>();
  10. 	/**条件等待与通知*/
  11.     private final Condition available = lock.newCondition();
  13.     /**入队操作*/
  14.     public boolean offer(E e) {
  15.         final ReentrantLock lock = this.lock;
  16.         lock.lock();
  17.         try {
  18.             q.offer(e);
  19.             if (q.peek() == e) {
  20.                 leader = null;
  21.                 available.signal();
  22.             }
  23.             return true;
  24.         } finally {
  25.             lock.unlock();
  26.         }
  27.     }
  29.     /**出队操作*/
  30.     public E poll() {
  31.         final ReentrantLock lock = this.lock;
  32.         lock.lock();
  33.         try {
  34.             E first = q.peek();
  35.             if (first == null || first.getDelay(NANOSECONDS) > 0)
  36.                 return null;
  37.             else
  38.                 return q.poll();
  39.         } finally {
  40.             lock.unlock();
  41.         }
  42.     }
  43. }

同样的，把最上面的样例Container中的阻塞队列实现类换成DelayQueue，代码如下：

  1. public class Container {
  3.     /**
  4.      * 初始化阻塞队列
  5.      */
  6.     private final BlockingQueue<DelayedUser> queue = new DelayQueue<DelayedUser>();
  9.     /**
  10.      * 添加数据到阻塞队列
  11.      * @param value
  12.      */
  13.     public void add(DelayedUser value) {
  14.         try {
  15.             queue.put(value);
  16.             System.out.println("生产者："+ Thread.currentThread().getName()+"，add：" + value);
  17.         } catch (InterruptedException e) {
  18.             e.printStackTrace();
  19.         }
  20.     }
  23.     /**
  24.      * 从阻塞队列获取数据
  25.      */
  26.     public void get() {
  27.         try {
  28.             DelayedUser value = queue.take();
  29.             String time = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
  30.             System.out.println(time + " 消费者："+ Thread.currentThread().getName()+"，value：" + value);
  31.         } catch (InterruptedException e) {
  32.             e.printStackTrace();
  33.         }
  34.     }
  35. }

DelayQueue队列中的元素需要显式实现Delayed接口，定义一个DelayedUser类，代码如下：

  1. public class DelayedUser implements Delayed {
  3.     /**
  4.      * 当前时间戳
  5.      */
  6.     private long start;
  8.     /**
  9.      * 延迟时间(单位：毫秒)
  10.      */
  11.     private long delayedTime;
  13.     /**
  14.      * 名称
  15.      */
  16.     private String name;
  18.     public DelayedUser(long delayedTime, String name) {
  19.         this.start = System.currentTimeMillis();
  20.         this.delayedTime = delayedTime;
  21.         this.name = name;
  22.     }
  24.     @Override
  25.     public long getDelay(TimeUnit unit) {
  26.         // 获取当前延迟的时间
  27.         long diffTime = (start + delayedTime) - System.currentTimeMillis();
  28.         return unit.convert(diffTime,TimeUnit.MILLISECONDS);
  29.     }
  31.     @Override
  32.     public int compareTo(Delayed o) {
  33.         // 判断当前对象的延迟时间是否大于目标对象的延迟时间
  34.         return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
  35.     }
  37.     @Override
  38.     public String toString() {
  39.         return "DelayedUser{" +
  40.                 "delayedTime=" + delayedTime +
  41.                 ", name='" + name + '\'' +
  42.                 '}';
  43.     }
  44. }

生产者插入数据的内容，做如下调整。

  1. /**
  2.  * 生产者
  3.  */
  4. public class Producer extends Thread {
  6.     private Container container;
  8.     public Producer(Container container) {
  9.         this.container = container;
  10.     }
  12.     @Override
  13.     public void run() {
  14.         for (int i = 0; i < 6; i++) {
  15.             container.add(new DelayedUser(1000 * i, "张三" +  i));
  16.         }
  17.     }
  18. }

最后运行结果如下：

  1. 生产者：Thread-0，add：DelayedUser{delayedTime=0, name='张三0'}
  2. 生产者：Thread-0，add：DelayedUser{delayedTime=1000, name='张三1'}
  3. 生产者：Thread-0，add：DelayedUser{delayedTime=2000, name='张三2'}
  4. 生产者：Thread-0，add：DelayedUser{delayedTime=3000, name='张三3'}
  5. 生产者：Thread-0，add：DelayedUser{delayedTime=4000, name='张三4'}
  6. 生产者：Thread-0，add：DelayedUser{delayedTime=5000, name='张三5'}
  7. 2023-11-03 14:55:33 消费者：Thread-1，value：DelayedUser{delayedTime=0, name='张三0'}
  8. 2023-11-03 14:55:34 消费者：Thread-1，value：DelayedUser{delayedTime=1000, name='张三1'}
  9. 2023-11-03 14:55:35 消费者：Thread-1，value：DelayedUser{delayedTime=2000, name='张三2'}
  10. 2023-11-03 14:55:36 消费者：Thread-1，value：DelayedUser{delayedTime=3000, name='张三3'}
  11. 2023-11-03 14:55:37 消费者：Thread-1，value：DelayedUser{delayedTime=4000, name='张三4'}
  12. 2023-11-03 14:55:38 消费者：Thread-1，value：DelayedUser{delayedTime=5000, name='张三5'}

可以很清晰的看到，延迟时间最低的排在最前面。

四、小结

最后我们来总结一下BlockingQueue阻塞队列接口，它提供了很多非常丰富的生产者和消费者模型的编程实现，同时兼顾了线程安全和执行效率的特点。

开发者可以通过BlockingQueue阻塞队列接口，简单的代码编程即可实现多线程中数据高效安全传输的目的，确切的说，它帮助开发者减轻了不少的编程难度。

在实际的业务开发中，其中LinkedBlockingQueue使用的是最广泛的，因为它的执行效率最高，在使用的时候，需要平衡好队列长度，防止过大导致内存溢出。

举个最简单的例子，比如某个功能上线之后，需要做下压力测试，总共需要请求 10000 次，采用 100 个线程去执行，测试服务是否能正常工作。如何实现呢？

可能有的同学想到，每个线程执行 100 次请求，启动 100 个线程去执行，可以是可以，就是有点笨拙。

其实还有另一个办法，就是将 10000 个请求对象，存入到阻塞队列中，然后采用 100 个线程去消费执行，这种编程模型会更佳灵活。

具体示例代码如下：

  1. public static void main(String[] args) throws InterruptedException {
  2.     // 将每个用户访问百度服务的请求任务，存入阻塞队列中
  3.     // 也可以也采用多线程写入
  4.     BlockingQueue<String> queue = new LinkedBlockingQueue<>();
  5.     for (int i = 0; i < 10000; i++) {
  6.         queue.put("https://www.baidu.com?paramKey=" + i);
  7.     }
  9.     // 模拟100个线程，执行10000次请求访问百度
  10.     final int threadNum = 100;
  11.     for (int i = 0; i < threadNum; i++) {
  12.         final int threadCount = i + 1;
  13.         new Thread(new Runnable() {
  15.             @Override
  16.             public void run() {
  17.                 System.out.println("thread " + threadCount + " start");
  18.                 boolean over = false;
  19.                 while (!over) {
  20.                     String url = queue.poll();
  21.                     if(Objects.nonNull(url)) {
  22.                         // 发起请求
  23.                         String result =HttpUtils.getUrl(url);
  24.                         System.out.println("thread " + threadCount + " run result:" + result);
  25.                     }else {
  26.                         // 任务结束
  27.                         over = true;
  28.                         System.out.println("thread " + threadCount + " final");
  29.                     }
  30.                 }
  31.             }
  32.         }).start();
  33.     }
  34. }

本文主要围绕BlockingQueue阻塞队列接口，从方法介绍到用法详解，做了一次知识总结，如果有描述不对的地方，欢迎留言指出！

五、参考

1、https://www.cnblogs.com/xrq730/p/4855857.html

2、https://juejin.cn/post/6999798721269465102

原文链接：https://www.cnblogs.com/dxflqm/p/18053297

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

Java热门文章

  • 【IDEA】Application context not configured for this file
  • [IDEA] chapter_reader - idea看小说插件 idea阅读插件 idea摸鱼插件
  • thrift常见异常及原因分析
  • Java 调用 PaddleDetection 模型
  • JVM 详解，大白话带你认识 JVM
  • spring-boot-2.0.3不一样系列之番外篇 - @Configuration、Condition与@Conditional
  • Java Comparator.comparing比较导致空指针异常的解决
  • Java使用EasyExcel进行单元格合并的问题详解
  • 死磕 java集合之ArrayBlockingQueue源码分析
  • Java基础之集合Set详解

    • Java推荐文章

  • 面试场景题：一次关于线程池使用场景的讨论。
  • 前端黑科技：使用 JavaScript 实现网页扫码功能
  • 聊一聊 Netty 数据搬运工 ByteBuf 体系的设计与实现
  • 简单设计一个JAVA并行处理工具类
  • 【杂谈】JPA乐观锁改悲观锁遇到的一些问题与思考
  • 认识netty的基本组件
  • Jenkins如何使用CrumbIssuer防御CSRF攻击
  • 如何平稳地从nacos迁移到r-nacos？
  • 线程池遇到父子任务，有大坑，要注意！
  • Java JVM——12. 垃圾回收理论概述

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号