当前位置:首页 > 科技  > 软件

一文带你彻底掌握阻塞队列!

来源: 责编: 时间:2023-12-15 09:49:51 140观看
导读一、摘要在之前的文章中,我们介绍了生产者和消费者模型的最基本实现思路,相信大家对它已经有一个初步的认识。在 Java 的并发包里面还有一个非常重要的接口:BlockingQueue。BlockingQueue是一个阻塞队列,更为准确的解释是

一、摘要

在之前的文章中,我们介绍了生产者和消费者模型的最基本实现思路,相信大家对它已经有一个初步的认识。Ldb28资讯网——每日最新资讯28at.com

在 Java 的并发包里面还有一个非常重要的接口:BlockingQueue。Ldb28资讯网——每日最新资讯28at.com

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

Ldb28资讯网——每日最新资讯28at.com

那什么是阻塞队列呢?Ldb28资讯网——每日最新资讯28at.com

简单的说,就是当参数在入队和出队时,通过加锁的方式来避免线程并发操作时导致的数据异常问题。Ldb28资讯网——每日最新资讯28at.com

在 Java 中,能对线程并发执行进行加锁的方式主要有synchronized和ReentrantLock,其中BlockingQueue采用的是ReentrantLock方式实现。Ldb28资讯网——每日最新资讯28at.com

与此对应的还有非阻塞机制的队列,主要是采用 CAS 方式来控制并发操作,例如:ConcurrentLinkedQueue,这个我们在后面的文章再进行分享介绍。Ldb28资讯网——每日最新资讯28at.com

今天我们主要介绍BlockingQueue相关的知识和用法,废话不多说了,进入正题!Ldb28资讯网——每日最新资讯28at.com

二、BlockingQueue 方法介绍

打开BlockingQueue的源码,你会发现它继承自Queue,正如上文提到的,它本质是一个队列接口。Ldb28资讯网——每日最新资讯28at.com

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

关于队列,我们在之前的集合系列文章中对此有过深入的介绍,本篇就再次简单的介绍一下。Ldb28资讯网——每日最新资讯28at.com

队列其实是一个数据结构,元素遵循先进先出的原则,所有新元素的插入,也被称为入队操作,会插入到队列的尾部;元素的移除,也被称为出队操作,会从队列的头部开始移除,从而保证先进先出的原则。Ldb28资讯网——每日最新资讯28at.com

在Queue接口中,总共有 6 个方法,可以分为 3 类,分别是:插入、移除、查询,内容如下:Ldb28资讯网——每日最新资讯28at.com

方法描述add(e)插入元素,如果插入失败,就抛异常offer(e)插入元素,如果插入成功,就返回 true;反之 falseremove()移除元素,如果移除失败,就抛异常poll()移除元素,如果移除成功,返回 true;反之 falseelement()获取队首元素,如果获取结果为空,就抛异常peek()获取队首元素,如果获取结果为空,返回空对象Ldb28资讯网——每日最新资讯28at.com

因为BlockingQueue是Queue的子接口,了解Queue接口里面的方法,有助于我们对BlockingQueue的理解。Ldb28资讯网——每日最新资讯28at.com

除此之外,BlockingQueue还单独扩展了一些特有的方法,内容如下:Ldb28资讯网——每日最新资讯28at.com

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

分析源码,你会发现相比普通的Queue子类,BlockingQueue子类主要有以下几个明显的不同点:Ldb28资讯网——每日最新资讯28at.com

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

三、BlockingQueue 用法详解

打开源码,BlockingQueue接口的实现类非常多,我们重点讲解一下其中的 5 个非常重要的实现类,分别如下表所示。Ldb28资讯网——每日最新资讯28at.com

实现类功能ArrayBlockingQueue基于数组的阻塞队列,使用数组存储数据,需要指定长度,所以是一个有界队列LinkedBlockingQueue基于链表的阻塞队列,使用链表存储数据,默认是一个无界队列;也可以通过构造方法中的capacity设置最大元素数量,所以也可以作为有界队列SynchronousQueue一种没有缓冲的队列Ldb28资讯网——每日最新资讯28at.com

生产者产生的数据直接会被消费者获取并且立刻消费PriorityBlockingQueue基于优先级别的阻塞队列,底层基于数组实现,是一个无界队列DelayQueue延迟队列,其中的元素只有到了其指定的延迟时间,才能够从队列中出队Ldb28资讯网——每日最新资讯28at.com

下面我们对以上实现类的用法,进行一一介绍。Ldb28资讯网——每日最新资讯28at.com

3.1、ArrayBlockingQueue

ArrayBlockingQueue是一个基于数组的阻塞队列,初始化的时候必须指定队列大小,源码实现比较简单,采用的是ReentrantLock和Condition实现生产者和消费者模型,部分核心源码如下:Ldb28资讯网——每日最新资讯28at.com

public class ArrayBlockingQueue<E> extends AbstractQueue<E>        implements BlockingQueue<E>, java.io.Serializable { /** 使用数组存储队列中的元素 */ final Object[] items; /** 使用独占锁ReetrantLock */ final ReentrantLock lock; /** 等待出队的条件 */ private final Condition notEmpty; /** 等待入队的条件 */ private final Condition notFull; /** 初始化时,需要指定队列大小 */ public ArrayBlockingQueue(int capacity) {        this(capacity, false);    }    /** 初始化时,也指出指定是否为公平锁, */    public ArrayBlockingQueue(int capacity, boolean fair) {        if (capacity <= 0)            throw new IllegalArgumentException();        this.items = new Object[capacity];        lock = new ReentrantLock(fair);        notEmpty = lock.newCondition();        notFull =  lock.newCondition();    }    /**入队操作*/    public void put(E e) throws InterruptedException {        checkNotNull(e);        final ReentrantLock lock = this.lock;        lock.lockInterruptibly();        try {            while (count == items.length)                notFull.await();            enqueue(e);        } finally {            lock.unlock();        }    }    /**出队操作*/    public E take() throws InterruptedException {        final ReentrantLock lock = this.lock;        lock.lockInterruptibly();        try {            while (count == 0)                notEmpty.await();            return dequeue();        } finally {            lock.unlock();        }    }}

ArrayBlockingQueue采用ReentrantLock进行加锁,只有一个ReentrantLock对象,这意味着生产者和消费者无法并行运行。Ldb28资讯网——每日最新资讯28at.com

我们看一个简单的示例代码如下:Ldb28资讯网——每日最新资讯28at.com

public class Container {    /**     * 初始化阻塞队列     */    private final BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);    /**     * 添加数据到阻塞队列     * @param value     */    public void add(Integer value) {        try {            queue.put(value);            System.out.println("生产者:"+ Thread.currentThread().getName()+",add:" + value);        } catch (InterruptedException e) {            e.printStackTrace();        }    }    /**     * 从阻塞队列获取数据     */    public void get() {        try {            Integer value = queue.take();            System.out.println("消费者:"+ Thread.currentThread().getName()+",value:" + value);        } catch (InterruptedException e) {            e.printStackTrace();        }    }}
/** * 生产者 */public class Producer extends Thread {    private Container container;    public Producer(Container container) {        this.container = container;    }    @Override    public void run() {        for (int i = 0; i < 6; i++) {            container.add(i);        }    }}
/** * 消费者 */public class Consumer extends Thread {    private Container container;    public Consumer(Container container) {        this.container = container;    }    @Override    public void run() {        for (int i = 0; i < 6; i++) {            container.get();        }    }}
/** * 测试类 */public class MyThreadTest {    public static void main(String[] args) {        Container container = new Container();        Producer producer = new Producer(container);        Consumer consumer = new Consumer(container);        producer.start();        consumer.start();    }}

运行结果如下:Ldb28资讯网——每日最新资讯28at.com

生产者:Thread-0,add:0生产者:Thread-0,add:1生产者:Thread-0,add:2生产者:Thread-0,add:3生产者:Thread-0,add:4生产者:Thread-0,add:5消费者:Thread-1,value:0消费者:Thread-1,value:1消费者:Thread-1,value:2消费者:Thread-1,value:3消费者:Thread-1,value:4消费者:Thread-1,value:5

可以很清晰的看到,生产者线程执行完毕之后,消费者线程才开始消费。Ldb28资讯网——每日最新资讯28at.com

3.2、LinkedBlockingQueue

LinkedBlockingQueue是一个基于链表的阻塞队列,初始化的时候无须指定队列大小,默认队列长度为Integer.MAX_VALUE,也就是 int 型最大值。Ldb28资讯网——每日最新资讯28at.com

同样的,采用的是ReentrantLock和Condition实现生产者和消费者模型,不同的是它使用了两个lock,这意味着生产者和消费者可以并行运行,程序执行效率进一步得到提升。Ldb28资讯网——每日最新资讯28at.com

部分核心源码如下:Ldb28资讯网——每日最新资讯28at.com

public class LinkedBlockingQueue<E> extends AbstractQueue<E>        implements BlockingQueue<E>, java.io.Serializable {    /** 使用出队独占锁ReetrantLock */    private final ReentrantLock takeLock = new ReentrantLock();    /** 等待出队的条件 */    private final Condition notEmpty = takeLock.newCondition();    /** 使用入队独占锁ReetrantLock */    private final ReentrantLock putLock = new ReentrantLock();    /** 等待入队的条件 */    private final Condition notFull = putLock.newCondition();    /**入队操作*/    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();    }    /**出队操作*/    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;    }}

把最上面的样例Container中的阻塞队列实现类换成LinkedBlockingQueue,调整如下:Ldb28资讯网——每日最新资讯28at.com

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

再次运行结果如下:Ldb28资讯网——每日最新资讯28at.com

生产者:Thread-0,add:0消费者:Thread-1,value:0生产者:Thread-0,add:1消费者:Thread-1,value:1生产者:Thread-0,add:2消费者:Thread-1,value:2生产者:Thread-0,add:3生产者:Thread-0,add:4生产者:Thread-0,add:5消费者:Thread-1,value:3消费者:Thread-1,value:4消费者:Thread-1,value:5

可以很清晰的看到,生产者线程和消费者线程,交替并行执行。Ldb28资讯网——每日最新资讯28at.com

3.3、SynchronousQueue

SynchronousQueue是一个没有缓冲的队列,生产者产生的数据直接会被消费者获取并且立刻消费,相当于传统的一个请求对应一个应答模式。Ldb28资讯网——每日最新资讯28at.com

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

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

部分核心源码如下:Ldb28资讯网——每日最新资讯28at.com

public class SynchronousQueue<E> extends AbstractQueue<E>    implements BlockingQueue<E>, java.io.Serializable {    /**不同的策略实现*/    private transient volatile Transferer<E> transferer; /**默认非公平模式*/    public SynchronousQueue() {        this(false);    }    /**可以选策略,也可以采用公平模式*/    public SynchronousQueue(boolean fair) {        transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();    } /**入队操作*/    public void put(E e) throws InterruptedException {        if (e == null) throw new NullPointerException();        if (transferer.transfer(e, false, 0) == null) {            Thread.interrupted();            throw new InterruptedException();        }    }    /**出队操作*/    public E take() throws InterruptedException {        E e = transferer.transfer(null, false, 0);        if (e != null)            return e;        Thread.interrupted();        throw new InterruptedException();    }}

同样的,把最上面的样例Container中的阻塞队列实现类换成SynchronousQueue,代码如下:Ldb28资讯网——每日最新资讯28at.com

public class Container {    /**     * 初始化阻塞队列     */    private final BlockingQueue<Integer> queue = new SynchronousQueue<>();    /**     * 添加数据到阻塞队列     * @param value     */    public void add(Integer value) {        try {            queue.put(value);            Thread.sleep(100);            System.out.println("生产者:"+ Thread.currentThread().getName()+",add:" + value);        } catch (InterruptedException e) {            e.printStackTrace();        }    }    /**     * 从阻塞队列获取数据     */    public void get() {        try {            Integer value = queue.take();            Thread.sleep(200);            System.out.println("消费者:"+ Thread.currentThread().getName()+",value:" + value);        } catch (InterruptedException e) {            e.printStackTrace();        }    }}

再次运行结果如下:Ldb28资讯网——每日最新资讯28at.com

生产者:Thread-0,add:0消费者:Thread-1,value:0生产者:Thread-0,add:1消费者:Thread-1,value:1生产者:Thread-0,add:2消费者:Thread-1,value:2生产者:Thread-0,add:3消费者:Thread-1,value:3生产者:Thread-0,add:4消费者:Thread-1,value:4生产者:Thread-0,add:5消费者:Thread-1,value:5

可以很清晰的看到,生产者线程和消费者线程,交替串行执行,生产者每投递一条数据,消费者处理一条数据。Ldb28资讯网——每日最新资讯28at.com

3.4、PriorityBlockingQueue

PriorityBlockingQueue是一个基于优先级别的阻塞队列,底层基于数组实现,可以认为是一个无界队列。Ldb28资讯网——每日最新资讯28at.com

PriorityBlockingQueue与ArrayBlockingQueue的实现逻辑,基本相似,也是采用ReentrantLock来实现加锁的操作。Ldb28资讯网——每日最新资讯28at.com

最大不同点在于:Ldb28资讯网——每日最新资讯28at.com

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

部分核心源码如下:Ldb28资讯网——每日最新资讯28at.com

public class PriorityBlockingQueue<E> extends AbstractQueue<E>    implements BlockingQueue<E>, java.io.Serializable {  /**队列元素*/    private transient Object[] queue;    /**比较器*/    private transient Comparator<? super E> comparator;    /**采用ReentrantLock进行加锁*/    private final ReentrantLock lock;    /**条件等待与通知*/    private final Condition notEmpty;    /**入队操作*/    public boolean offer(E e) {        if (e == null)            throw new NullPointerException();        final ReentrantLock lock = this.lock;        lock.lock();        int n, cap;        Object[] array;        while ((n = size) >= (cap = (array = queue).length))            tryGrow(array, cap);        try {            Comparator<? super E> cmp = comparator;            if (cmp == null)                siftUpComparable(n, e, array);            else                siftUpUsingComparator(n, e, array, cmp);            size = n + 1;            notEmpty.signal();        } finally {            lock.unlock();        }        return true;    }    /**出队操作*/    public E take() throws InterruptedException {        final ReentrantLock lock = this.lock;        lock.lockInterruptibly();        E result;        try {            while ( (result = dequeue()) == null)                notEmpty.await();        } finally {            lock.unlock();        }        return result;    }}

同样的,把最上面的样例Container中的阻塞队列实现类换成PriorityBlockingQueue,调整如下:Ldb28资讯网——每日最新资讯28at.com

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

生产者插入数据的内容,我们改下插入顺序。Ldb28资讯网——每日最新资讯28at.com

/** * 生产者 */public class Producer extends Thread {    private Container container;    public Producer(Container container) {        this.container = container;    }    @Override    public void run() {        container.add(5);        container.add(3);        container.add(1);        container.add(2);        container.add(0);        container.add(4);    }}

最后运行结果如下:Ldb28资讯网——每日最新资讯28at.com

生产者:Thread-0,add:5生产者:Thread-0,add:3生产者:Thread-0,add:1生产者:Thread-0,add:2生产者:Thread-0,add:0生产者:Thread-0,add:4消费者:Thread-1,value:0消费者:Thread-1,value:1消费者:Thread-1,value:2消费者:Thread-1,value:3消费者:Thread-1,value:4消费者:Thread-1,value:5

从日志上可以很明显看出,对于整数,默认情况下,按照升序排序,消费者默认从 0 开始处理。Ldb28资讯网——每日最新资讯28at.com

3.5、DelayQueue

DelayQueue是一个线程安全的延迟队列,存入队列的元素不会立刻被消费,只有到了其指定的延迟时间,才能够从队列中出队。Ldb28资讯网——每日最新资讯28at.com

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

部分核心源码如下:Ldb28资讯网——每日最新资讯28at.com

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>    implements BlockingQueue<E> {    /**采用ReentrantLock进行加锁*/    private final transient ReentrantLock lock = new ReentrantLock();    /**采用PriorityQueue进行存储数据*/    private final PriorityQueue<E> q = new PriorityQueue<E>(); /**条件等待与通知*/    private final Condition available = lock.newCondition();    /**入队操作*/    public boolean offer(E e) {        final ReentrantLock lock = this.lock;        lock.lock();        try {            q.offer(e);            if (q.peek() == e) {                leader = null;                available.signal();            }            return true;        } finally {            lock.unlock();        }    }    /**出队操作*/    public E poll() {        final ReentrantLock lock = this.lock;        lock.lock();        try {            E first = q.peek();            if (first == null || first.getDelay(NANOSECONDS) > 0)                return null;            else                return q.poll();        } finally {            lock.unlock();        }    }}

同样的,把最上面的样例Container中的阻塞队列实现类换成DelayQueue,代码如下:Ldb28资讯网——每日最新资讯28at.com

public class Container {    /**     * 初始化阻塞队列     */    private final BlockingQueue<DelayedUser> queue = new DelayQueue<DelayedUser>();    /**     * 添加数据到阻塞队列     * @param value     */    public void add(DelayedUser value) {        try {            queue.put(value);            System.out.println("生产者:"+ Thread.currentThread().getName()+",add:" + value);        } catch (InterruptedException e) {            e.printStackTrace();        }    }    /**     * 从阻塞队列获取数据     */    public void get() {        try {            DelayedUser value = queue.take();            String time = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());            System.out.println(time + " 消费者:"+ Thread.currentThread().getName()+",value:" + value);        } catch (InterruptedException e) {            e.printStackTrace();        }    }}

DelayQueue队列中的元素需要显式实现Delayed接口,定义一个DelayedUser类,代码如下:Ldb28资讯网——每日最新资讯28at.com

public class DelayedUser implements Delayed {    /**     * 当前时间戳     */    private long start;    /**     * 延迟时间(单位:毫秒)     */    private long delayedTime;    /**     * 名称     */    private String name;    public DelayedUser(long delayedTime, String name) {        this.start = System.currentTimeMillis();        this.delayedTime = delayedTime;        this.name = name;    }    @Override    public long getDelay(TimeUnit unit) {        // 获取当前延迟的时间        long diffTime = (start + delayedTime) - System.currentTimeMillis();        return unit.convert(diffTime,TimeUnit.MILLISECONDS);    }    @Override    public int compareTo(Delayed o) {        // 判断当前对象的延迟时间是否大于目标对象的延迟时间        return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));    }    @Override    public String toString() {        return "DelayedUser{" +                "delayedTime=" + delayedTime +                ", name='" + name + '/'' +                '}';    }}

生产者插入数据的内容,做如下调整。Ldb28资讯网——每日最新资讯28at.com

/** * 生产者 */public class Producer extends Thread {    private Container container;    public Producer(Container container) {        this.container = container;    }    @Override    public void run() {        for (int i = 0; i < 6; i++) {            container.add(new DelayedUser(1000 * i, "张三" +  i));        }    }}

最后运行结果如下:Ldb28资讯网——每日最新资讯28at.com

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

可以很清晰的看到,延迟时间最低的排在最前面。Ldb28资讯网——每日最新资讯28at.com

四、小结

最后我们来总结一下BlockingQueue阻塞队列接口,它提供了很多非常丰富的生产者和消费者模型的编程实现,同时兼顾了线程安全和执行效率的特点。Ldb28资讯网——每日最新资讯28at.com

开发者可以通过BlockingQueue阻塞队列接口,简单的代码编程即可实现多线程中数据高效安全传输的目的,确切的说,它帮助开发者减轻了不少的编程难度。Ldb28资讯网——每日最新资讯28at.com

在实际的业务开发中,其中LinkedBlockingQueue使用的是最广泛的,因为它的执行效率最高,在使用的时候,需要平衡好队列长度,防止过大导致内存溢出。Ldb28资讯网——每日最新资讯28at.com

举个最简单的例子,比如某个功能上线之后,需要做下压力测试,总共需要请求 10000 次,采用 100 个线程去执行,测试服务是否能正常工作。如何实现呢?Ldb28资讯网——每日最新资讯28at.com

可能有的同学想到,每个线程执行 100 次请求,启动 100 个线程去执行,可以是可以,就是有点笨拙。Ldb28资讯网——每日最新资讯28at.com

其实还有另一个办法,就是将 10000 个请求对象,存入到阻塞队列中,然后采用 100 个线程去消费执行,这种编程模型会更佳灵活。Ldb28资讯网——每日最新资讯28at.com

具体示例代码如下:Ldb28资讯网——每日最新资讯28at.com

public static void main(String[] args) throws InterruptedException {    // 将每个用户访问百度服务的请求任务,存入阻塞队列中    // 也可以也采用多线程写入    BlockingQueue<String> queue = new LinkedBlockingQueue<>();    for (int i = 0; i < 10000; i++) {        queue.put("https://www.baidu.com?paramKey=" + i);    }    // 模拟100个线程,执行10000次请求访问百度    final int threadNum = 100;    for (int i = 0; i < threadNum; i++) {        final int threadCount = i + 1;        new Thread(new Runnable() {            @Override            public void run() {                System.out.println("thread " + threadCount + " start");                boolean over = false;                while (!over) {                    String url = queue.poll();                    if(Objects.nonNull(url)) {                        // 发起请求                        String result =HttpUtils.getUrl(url);                        System.out.println("thread " + threadCount + " run result:" + result);                    }else {                        // 任务结束                        over = true;                        System.out.println("thread " + threadCount + " final");                    }                }            }        }).start();    }}

本文主要围绕BlockingQueue阻塞队列接口,从方法介绍到用法详解,做了一次知识总结,如果有描述不对的地方,欢迎留言指出!Ldb28资讯网——每日最新资讯28at.com

本文链接:http://www.28at.com/showinfo-26-46337-0.html一文带你彻底掌握阻塞队列!

声明:本网页内容旨在传播知识,若有侵权等问题请及时与本网联系,我们将在第一时间删除处理。邮件:2376512515@qq.com

上一篇: 格力董明珠再评孟羽童:只想着利用平台当网红,在公司内产生不好影响

下一篇: 7k Star,一款开源的 Kafka 管理平台,功能齐全、页面美观!

标签:
  • 热门焦点
  • 俄罗斯:将审查iPhone等外国公司设备 保数据安全

    俄罗斯:将审查iPhone等外国公司设备 保数据安全

    iPhone和特斯拉都属于在各自领域领头羊的品牌,推出的产品也也都是数一数二的,但对于一些国家而言,它们的产品可靠性和安全性还是在限制范围内。近日,俄罗斯联邦通信、信息技术
  • 7月安卓手机性价比榜:努比亚+红魔两款新机入榜

    7月安卓手机性价比榜:努比亚+红魔两款新机入榜

    7月登场的新机有努比亚Z50S Pro和红魔8S Pro,除了三星之外目前唯二的两款搭载超频版骁龙8Gen2处理器的产品,而且努比亚和红魔也一贯有着不错的性价比,所以在本次的性价比榜单
  • 6月iOS设备性能榜:M2稳居榜首 A系列只能等一手3nm来救

    6月iOS设备性能榜:M2稳居榜首 A系列只能等一手3nm来救

    没有新品发布,自然iOS设备性能榜的上榜设备就没有什么更替,仅仅只有跑分变化而产生的排名变动,毕竟苹果新品的发布节奏就是这样的,一年下来也就几个移动端新品,不会像安卓厂商,一
  • 从 Pulsar Client 的原理到它的监控面板

    从 Pulsar Client 的原理到它的监控面板

    背景前段时间业务团队偶尔会碰到一些 Pulsar 使用的问题,比如消息阻塞不消费了、生产者消息发送缓慢等各种问题。虽然我们有个监控页面可以根据 topic 维度查看他的发送状态,
  • 微信语音大揭秘:为什么禁止转发?

    微信语音大揭秘:为什么禁止转发?

    大家好,我是你们的小米。今天,我要和大家聊一个有趣的话题:为什么微信语音不可以转发?这是一个我们经常在日常使用中遇到的问题,也是一个让很多人好奇的问题。让我们一起来揭开这
  • 在线图片编辑器,支持PSD解析、AI抠图等

    在线图片编辑器,支持PSD解析、AI抠图等

    自从我上次分享一个人开发仿造稿定设计的图片编辑器到现在,不知不觉已过去一年时间了,期间我经历了裁员失业、面试找工作碰壁,寒冬下一直没有很好地履行计划.....这些就放在日
  • 三分钟白话RocketMQ系列—— 如何发送消息

    三分钟白话RocketMQ系列—— 如何发送消息

    我们知道RocketMQ主要分为消息 生产、存储(消息堆积)、消费 三大块领域。那接下来,我们白话一下,RocketMQ是如何发送消息的,揭秘消息生产全过程。注意,如果白话中不小心提到相关代
  • 2299元起!iQOO Pad明晚首销:性能最强天玑平板

    2299元起!iQOO Pad明晚首销:性能最强天玑平板

    5月23日,iQOO如期举行了新品发布会,除了首发安卓最强旗舰处理器的iQOO Neo8系列新机外,还在发布会上推出了旗下首款平板电脑——iQOO Pad,其最大的卖点
  • 荣耀Magic4 至臻版 首创智慧隐私通话 强劲影音系统

    荣耀Magic4 至臻版 首创智慧隐私通话 强劲影音系统

    2022年第一季度临近尾声,在该季度内,许多品牌陆续发布自己的最新产品,让大家从全新的角度来了解当今的手机技术。手机是电子设备中,更新迭代十分迅速的一款产品,基
Top