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

一、摘要

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

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

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

那什么是阻塞队列呢?

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

在 Java 中,能对线程并发执行进行加锁的方式主要有synchronizedReentrantLock,其中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() 获取队首元素,如果获取结果为空,返回空对象

因为BlockingQueueQueue的子接口,了解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是一个基于数组的阻塞队列,初始化的时候必须指定队列大小,源码实现比较简单,采用的是ReentrantLockCondition实现生产者和消费者模型,部分核心源码如下:

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

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

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

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

运行结果如下:

  1. 生产者:Thread-0add0
  2. 生产者:Thread-0add1
  3. 生产者:Thread-0add2
  4. 生产者:Thread-0add3
  5. 生产者:Thread-0add4
  6. 生产者:Thread-0add5
  7. 消费者:Thread-1value0
  8. 消费者:Thread-1value1
  9. 消费者:Thread-1value2
  10. 消费者:Thread-1value3
  11. 消费者:Thread-1value4
  12. 消费者:Thread-1value5

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

3.2、LinkedBlockingQueue

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

同样的,采用的是ReentrantLockCondition实现生产者和消费者模型,不同的是它使用了两个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();
  5. /** 等待出队的条件 */
  6. private final Condition notEmpty = takeLock.newCondition();
  7. /** 使用入队独占锁ReetrantLock */
  8. private final ReentrantLock putLock = new ReentrantLock();
  9. /** 等待入队的条件 */
  10. private final Condition notFull = putLock.newCondition();
  11. /**入队操作*/
  12. public void put(E e) throws InterruptedException {
  13. if (e == null) throw new NullPointerException();
  14. int c = -1;
  15. Node<E> node = new Node<E>(e);
  16. final ReentrantLock putLock = this.putLock;
  17. final AtomicInteger count = this.count;
  18. putLock.lockInterruptibly();
  19. try {
  20. while (count.get() == capacity) {
  21. notFull.await();
  22. }
  23. enqueue(node);
  24. c = count.getAndIncrement();
  25. if (c + 1 < capacity)
  26. notFull.signal();
  27. } finally {
  28. putLock.unlock();
  29. }
  30. if (c == 0)
  31. signalNotEmpty();
  32. }
  33. /**出队操作*/
  34. public E take() throws InterruptedException {
  35. E x;
  36. int c = -1;
  37. final AtomicInteger count = this.count;
  38. final ReentrantLock takeLock = this.takeLock;
  39. takeLock.lockInterruptibly();
  40. try {
  41. while (count.get() == 0) {
  42. notEmpty.await();
  43. }
  44. x = dequeue();
  45. c = count.getAndDecrement();
  46. if (c > 1)
  47. notEmpty.signal();
  48. } finally {
  49. takeLock.unlock();
  50. }
  51. if (c == capacity)
  52. signalNotFull();
  53. return x;
  54. }
  55. }

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

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

再次运行结果如下:

  1. 生产者:Thread-0add0
  2. 消费者:Thread-1value0
  3. 生产者:Thread-0add1
  4. 消费者:Thread-1value1
  5. 生产者:Thread-0add2
  6. 消费者:Thread-1value2
  7. 生产者:Thread-0add3
  8. 生产者:Thread-0add4
  9. 生产者:Thread-0add5
  10. 消费者:Thread-1value3
  11. 消费者:Thread-1value4
  12. 消费者:Thread-1value5

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

3.3、SynchronousQueue

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

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

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

部分核心源码如下:

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

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

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

再次运行结果如下:

  1. 生产者:Thread-0add0
  2. 消费者:Thread-1value0
  3. 生产者:Thread-0add1
  4. 消费者:Thread-1value1
  5. 生产者:Thread-0add2
  6. 消费者:Thread-1value2
  7. 生产者:Thread-0add3
  8. 消费者:Thread-1value3
  9. 生产者:Thread-0add4
  10. 消费者:Thread-1value4
  11. 生产者:Thread-0add5
  12. 消费者:Thread-1value5

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

3.4、PriorityBlockingQueue

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

PriorityBlockingQueueArrayBlockingQueue的实现逻辑,基本相似,也是采用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 {
  3. /**队列元素*/
  4. private transient Object[] queue;
  5. /**比较器*/
  6. private transient Comparator<? super E> comparator;
  7. /**采用ReentrantLock进行加锁*/
  8. private final ReentrantLock lock;
  9. /**条件等待与通知*/
  10. private final Condition notEmpty;
  11. /**入队操作*/
  12. public boolean offer(E e) {
  13. if (e == null)
  14. throw new NullPointerException();
  15. final ReentrantLock lock = this.lock;
  16. lock.lock();
  17. int n, cap;
  18. Object[] array;
  19. while ((n = size) >= (cap = (array = queue).length))
  20. tryGrow(array, cap);
  21. try {
  22. Comparator<? super E> cmp = comparator;
  23. if (cmp == null)
  24. siftUpComparable(n, e, array);
  25. else
  26. siftUpUsingComparator(n, e, array, cmp);
  27. size = n + 1;
  28. notEmpty.signal();
  29. } finally {
  30. lock.unlock();
  31. }
  32. return true;
  33. }
  34. /**出队操作*/
  35. public E take() throws InterruptedException {
  36. final ReentrantLock lock = this.lock;
  37. lock.lockInterruptibly();
  38. E result;
  39. try {
  40. while ( (result = dequeue()) == null)
  41. notEmpty.await();
  42. } finally {
  43. lock.unlock();
  44. }
  45. return result;
  46. }
  47. }

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

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

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

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

最后运行结果如下:

  1. 生产者:Thread-0add5
  2. 生产者:Thread-0add3
  3. 生产者:Thread-0add1
  4. 生产者:Thread-0add2
  5. 生产者:Thread-0add0
  6. 生产者:Thread-0add4
  7. 消费者:Thread-1value0
  8. 消费者:Thread-1value1
  9. 消费者:Thread-1value2
  10. 消费者:Thread-1value3
  11. 消费者:Thread-1value4
  12. 消费者:Thread-1value5

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

3.5、DelayQueue

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

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

部分核心源码如下:

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

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

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

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

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

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

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

最后运行结果如下:

  1. 生产者:Thread-0addDelayedUser{delayedTime=0, name='张三0'}
  2. 生产者:Thread-0addDelayedUser{delayedTime=1000, name='张三1'}
  3. 生产者:Thread-0addDelayedUser{delayedTime=2000, name='张三2'}
  4. 生产者:Thread-0addDelayedUser{delayedTime=3000, name='张三3'}
  5. 生产者:Thread-0addDelayedUser{delayedTime=4000, name='张三4'}
  6. 生产者:Thread-0addDelayedUser{delayedTime=5000, name='张三5'}
  7. 2023-11-03 14:55:33 消费者:Thread-1valueDelayedUser{delayedTime=0, name='张三0'}
  8. 2023-11-03 14:55:34 消费者:Thread-1valueDelayedUser{delayedTime=1000, name='张三1'}
  9. 2023-11-03 14:55:35 消费者:Thread-1valueDelayedUser{delayedTime=2000, name='张三2'}
  10. 2023-11-03 14:55:36 消费者:Thread-1valueDelayedUser{delayedTime=3000, name='张三3'}
  11. 2023-11-03 14:55:37 消费者:Thread-1valueDelayedUser{delayedTime=4000, name='张三4'}
  12. 2023-11-03 14:55:38 消费者:Thread-1valueDelayedUser{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. }
  8. // 模拟100个线程,执行10000次请求访问百度
  9. final int threadNum = 100;
  10. for (int i = 0; i < threadNum; i++) {
  11. final int threadCount = i + 1;
  12. new Thread(new Runnable() {
  13. @Override
  14. public void run() {
  15. System.out.println("thread " + threadCount + " start");
  16. boolean over = false;
  17. while (!over) {
  18. String url = queue.poll();
  19. if(Objects.nonNull(url)) {
  20. // 发起请求
  21. String result =HttpUtils.getUrl(url);
  22. System.out.println("thread " + threadCount + " run result:" + result);
  23. }else {
  24. // 任务结束
  25. over = true;
  26. System.out.println("thread " + threadCount + " final");
  27. }
  28. }
  29. }
  30. }).start();
  31. }
  32. }

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

五、参考

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

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

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

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

本站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号