【Java数据结构及算法实战】系列009:Java队列03——数组实现的阻塞队列ArrayBlockingQueue

原创
04/27 21:55
阅读数 106

顾名思义,ArrayBlockingQueue是基于数组实现的有界阻塞队列。该队列对元素进行FIFO排序。队列的首元素是在该队列中驻留时间最长的元素。队列的尾部是在该队列中停留时间最短的元素。新的元素被插入到队列的尾部,队列检索操作获取队列头部的元素。

 

 

 

ArrayBlockingQueue是一个经典的“有界缓冲区(bounded buffer)”,其中内部包含了一个固定大小的数组,用于承载包含生产者插入的和消费者提取的元素。ArrayBlockingQueue的容量一旦创建,不可更改。试图将一个元素放入一个满队列将导致操作阻塞;试图从空队列中取出一个元素也同样会阻塞。

 

 

 

ArrayBlockingQueue支持排序的可选公平策略,用于等待生产者和消费者线程。默认情况下,不保证此顺序。然而,一个由公平性设置为true构造的队列允许线程以FIFO顺序访问。公平性一般会降低吞吐量,但可以减少可变性,避免线程饿死。

 

 

 

ArrayBlockingQueue类及其迭代器实现了CollectionIterator接口的所有可选方法。ArrayBlockingQueueJava Collections Framework的一个成员。

 

 

 

 

 

1.   ArrayBlockingQueue的声明

 

ArrayBlockingQueue的接口和继承关系如下

 

 

 

public class ArrayBlockingQueue<E> extends AbstractQueue<E>

 

        implements BlockingQueue<E>, java.io.Serializable {

 

 

 

   …

 

}

 

 

 

 

 

完整的接口继承关系如下图所示。

 

 

 

 

 

 

 

 

 

 

 

 

从上述代码可以看出,ArrayBlockingQueue既实现了BlockingQueue<E>java.io.Serializable接口,又继承了java.util.AbstractQueue<E>。其中,AbstractQueueQueue接口的抽象类,核心代码如下。

 

 

 

 

 

package java.util;

 

 

 

public abstract class AbstractQueue<E>

 

    extends AbstractCollection<E>

 

    implements Queue<E> {

 

 

 

    protected AbstractQueue() {

 

    }

 

 

 

    public boolean add(E e) {

 

        if (offer(e))

 

            return true;

 

        else

 

            throw new IllegalStateException("Queue full");

 

    }

 

 

 

    public E remove() {

 

        E x = poll();

 

        if (x != null)

 

            return x;

 

        else

 

            throw new NoSuchElementException();

 

    }

 

 

 

    public E element() {

 

        E x = peek();

 

        if (x != null)

 

            return x;

 

        else

 

            throw new NoSuchElementException();

 

    }

 

 

 

    public void clear() {

 

        while (poll() != null)

 

            ;

 

    }

 

 

 

    public boolean addAll(Collection<? extends E> c) {

 

        if (c == null)

 

            throw new NullPointerException();

 

        if (c == this)

 

            throw new IllegalArgumentException();

 

        boolean modified = false;

 

        for (E e : c)

 

            if (add(e))

 

                modified = true;

 

        return modified;

 

    }

 

 

 

}

 

 

 

 

 

2.   ArrayBlockingQueue的成员变量和构造函数

 

 

 

 

 

以下是ArrayBlockingQueue的构造函数和成员变量。

 

 

 

    // 元素数组

 

    final Object[] items;

 

 

 

    // 消费索引,用于takepollpeekremove操作

 

    int takeIndex;

 

 

 

    // 生产索引,用于putofferadd操作

 

    int putIndex;

 

 

 

    // 队列中的元素个数

 

    int count;

 

 

 

    /*

 

     * 使用经典的双条件算法(two-condition algorithm)实现并发控制

 

     */

 

 

 

    // 操作数组确保原子性的锁

 

    final ReentrantLock lock;

 

 

 

    // 数组非空,唤醒消费者

 

    private final Condition notEmpty;

 

 

 

    // 数组非满,唤醒生产者

 

private final Condition notFull;

 

 

 

// 迭代器状态

 

transient Itrs itrs;

 

 

 

public ArrayBlockingQueue(int capacity) {

 

        this(capacityfalse);

 

    }

 

 

 

    public ArrayBlockingQueue(int capacityboolean fair) {

 

        if (capacity <= 0)

 

            throw new IllegalArgumentException();

 

        this.items = new Object[capacity];

 

        lock = new ReentrantLock(fair);

 

        notEmpty = lock.newCondition();

 

        notFull =  lock.newCondition();

 

    }

 

 

 

    public ArrayBlockingQueue(int capacityboolean fair,

 

                              Collection<? extends E> c) {

 

        this(capacityfair);

 

 

 

        final ReentrantLock lock = this.lock;

 

        lock.lock(); // 只锁可见,不互斥

 

        try {

 

            final Object[] items = this.items;

 

            int i = 0;

 

            try {

 

                for (E e : c)

 

                    items[i++] = Objects.requireNonNull(e);

 

            } catch (ArrayIndexOutOfBoundsException ex) {

 

                throw new IllegalArgumentException();

 

            }

 

            count = i;

 

            putIndex = (i == capacity) ? 0 : i;

 

        } finally {

 

            lock.unlock();  // 解锁

 

        }

 

}

 

 

 

 

 

从上述代码可以看出,构造函数有三种。构造函数中的参数含义如下

 

 

 

l  capacity用于设置队列容量

 

l  fair用于设置访问策略,如果为true,则对线程的访问在插入或移除时被阻塞,则按FIFO顺序处理;如果为false,则访问顺序未指定

 

l  c用于设置最初包含给定集合的元素,按集合迭代器的遍历顺序添加

 

 

 

类成员items是一个数组,用于存储队列中的元素。关键字final指明了,当ArrayBlockingQueue构造完成之后,通过new Object[capacity]的方式初始化items数组完成后,则后续items的容量将不再变化。

 

 

 

访问策略是通过ReentrantLock来实现的。通过两个加锁条件notEmptynotFull来实现并发控制。这是典型的双条件算法(two-condition algorithm)。

 

 

 

ArrayBlockingQueue生产则增加putIndex,消费则增加takeIndex

 

 

 

Itrs用于记录当前活动迭代器的共享状态,如果已知不存在任何迭代器,则为null 允许队列操作更新迭代器状态。迭代器状态不是本节的重点,不再深入探讨。

 

3.   ArrayBlockingQueue的核心方法

 

以下对ArrayBlockingQueue常用核心方法的实现原理进行解释。

 

 

 

 

 

3.1.     offer(e)

 

执行offer(e)方法后有两种结果

 

 

 

l  队列未满时,返回 true

 

l  队列满时,返回 false

 

 

 

ArrayBlockingQueueoffer (e)方法源码如下:

 

 

 

public boolean offer(E e) {

 

        Objects.requireNonNull(e);

 

        final ReentrantLock lock = this.lock;

 

        lock.lock(); // 加锁

 

        try {

 

            if (count == items.length)

 

                return false;

 

            else {

 

                enqueue(e); // 入队

 

                return true;

 

            }

 

        } finally {

 

            lock.unlock();  // 解锁

 

        }

 

}

 

 

 

 

 

从上面代码可以看出,执行offer(e)方法时,分为以下几个步骤:

 

 

 

l  为了确保并发操作的安全先做了加锁处理。

 

l  而后判断count是否与数组items的长度一致,如果一致则证明队列已经满了,直接返回false;否则执行enqueue(e)方法做元素的入队,并返回true

 

l  最后解锁。

 

 

 

enqueue(e)方法源码如下:

 

 

 

 

 

private void enqueue(E e) {

 

        final Object[] items = this.items;

 

        items[putIndex] = e;

 

        if (++putIndex == items.lengthputIndex = 0;

 

        count++;

 

        notEmpty.signal(); // 唤醒等待中的线程

 

}

 

 

 

上面代码比较简单,在当前索引(putIndex)位置放置待入队的元素,而后putIndexcount分别递增,并通过signal()方法唤醒等待中的线程。其中一个注意点是,当putIndex 等于数组items长度时,putIndex置为0

 

 

 

思考:当putIndex 等于数组items长度时,putIndex为什么置为0呢?

 

 

 

3.2.     put(e)

 

执行put(e)方法后有两种结果:

 

      

 

l  队列未满时,直接插入没有返回值

 

l  队列满时,会阻塞等待,一直等到队列未满时再插入

 

 

 

ArrayBlockingQueueput (e)方法源码如下:

 

 

 

public void put(E ethrows InterruptedException {

 

        Objects.requireNonNull(e);

 

        final ReentrantLock lock = this.lock;

 

        lock.lockInterruptibly();  // 获取锁

 

        try {

 

            while (count == items.length)

 

                notFull.await();  // 使线程等待

 

            enqueue(e);  // 入队

 

        } finally {

 

            lock.unlock();  // 解锁

 

        }

 

    }

 

 

 

从上面代码可以看出,put(e)方法的实现,分为以下几个步骤:

 

 

 

l  先是要获取锁。

 

l  而后判断count是否与数组items的长度一致,如果一致则证明队列已经满了,就等待;否则执行enqueue(e)方法做元素的入队。

 

l  最后解锁。

 

 

 

3.3.     offer(e,time,unit)

 

offer(e,time,unit)方法与offer(e)方法不同之处在于,前者加入了等待机制。设定等待的时间,如果在指定时间内还不能往队列中插入数据则返回false。执行offer(e,time,unit)方法有两种结果:

 

      

 

l  队列未满时,返回 true

 

l  队列满时,会阻塞等待,如果在指定时间内还不能往队列中插入数据则返回 false

 

 

 

ArrayBlockingQueueput (e)方法源码如下:

 

 

 

public boolean offer(E elong timeout, TimeUnit unit)

 

        throws InterruptedException {

 

 

 

        Objects.requireNonNull(e);

 

        long nanos = unit.toNanos(timeout);

 

        final ReentrantLock lock = this.lock;

 

        lock.lockInterruptibly();  // 获取锁

 

        try {

 

            while (count == items.length) {

 

                if (nanos <= 0L)

 

                    return false;

 

                nanos = notFull.awaitNanos(nanos);  // 使线程等待指定的时间

 

            }

 

            enqueue(e);

 

            return true;

 

        } finally {

 

            lock.unlock();  // 解锁

 

        }

 

    }

 

 

 

从上面代码可以看出,offer(e,time,unit)方法的实现,分为以下几个步骤:

 

 

 

l  先是要获取锁。

 

l  而后判断count是否与数组items的长度一致,如果一致则证明队列已经满了,就等待;否则执行enqueue(e)方法做元素的入队。

 

l  最后解锁。

 

 

 

3.4.     add(e)

 

执行add(e)方法后有两种结果

 

 

 

l  队列未满时,返回 true

 

l  队列满时,则抛出异常

 

 

 

ArrayBlockingQueueadd(e)方法源码如下:

 

 

 

    public boolean add(E e) {

 

        return super.add(e);

 

    }

 

 

 

 

 

从上面代码可以看出,add(e)方法的实现,直接是调用了父类AbstractQueueadd(e)方法。而AbstractQueueadd(e)方法源码如下:

 

 

 

 

 

public boolean add(E e) {

 

        if (offer(e))

 

            return true;

 

        else

 

            throw new IllegalStateException("Queue full");

 

}

 

 

 

 

 

从上面代码可以看出,add(e)方法又调用了offer(e)方法。offer(e)方法此处不再赘述。

 

 

 

 

 

 

 

3.5.     poll ()

 

执行poll ()方法后有两种结果:

 

 

 

l  队列不为空时,返回队首值并移除

 

l  队列为空时,返回 null

 

 

 

 

 

ArrayBlockingQueuepoll()方法源码如下:

 

 

 

public E poll() {

 

        final ReentrantLock lock = this.lock;

 

        lock.lock();  // 加锁

 

        try {

 

            return (count == 0) ? null : dequeue(); // 出队

 

        } finally {

 

            lock.unlock();  // 解锁

 

        }

 

    }

 

 

 

从上面代码可以看出,执行poll()方法时,分为以下几个步骤:

 

 

 

l  为了确保并发操作的安全先做了加锁处理。

 

l  而后判断count是否等于0,如果等于0则证明队列为空,直接返回null;否则执行dequeue()方法做元素的出队。

 

l  最后解锁。

 

 

 

dequeue()方法源码如下:

 

 

 

 

 

private E dequeue() {

 

        final Object[] items = this.items;

 

        @SuppressWarnings("unchecked")

 

        E e = (E) items[takeIndex];

 

        items[takeIndex] = null;  // 删除数据

 

        if (++takeIndex == items.lengthtakeIndex = 0;

 

        count--;

 

        if (itrs != null)

 

            itrs.elementDequeued();

 

        notFull.signal(); // 唤醒等待中的线程

 

        return e;

 

}

 

 

 

上面代码比较简单,在当前索引(takeIndex)位置取出待出队的元素并删除队列中的元素,而后takeIndex递增count递减,并通过signal()方法唤醒等待中的线程。其中一个注意点是,当takeIndex等于数组items长度时,takeIndex置为0

 

 

 

3.6.     take()

 

执行take()方法后有两种结果:

 

 

 

l  队列不为空时,返回队首值并移除

 

l  队列为空时,会阻塞等待,一直等到队列不为空时再返回队首值

 

 

 

ArrayBlockingQueuetake ()方法源码如下:

 

 

 

public E take() throws InterruptedException {

 

        final ReentrantLock lock = this.lock;

 

        lock.lockInterruptibly();  // 获取锁

 

        try {

 

            while (count == 0)

 

                notEmpty.await(); // 使线程等待

 

            return dequeue();  // 出队

 

        } finally {

 

            lock.unlock();  // 解锁

 

        }

 

    }

 

 

 

从上面代码可以看出,执行take()方法时,分为以下几个步骤:

 

 

 

l  先是要获取锁。

 

l  而后判断count是否等于0,如果等于0则证明队列为空,会阻塞等待;否则执行dequeue()方法做元素的出队。

 

l  最后解锁。

 

 

 

dequeue()方法此处不再赘述。

 

 

 

3.7.     poll(time,unit)

 

poll(time,unit)方法与poll()方法不同之处在于,前者加入了等待机制。设定等待的时间,如果在指定时间内队列还为空,则返回null。执行poll(time,unit)方法后有两种结果:

 

 

 

l  队列不为空时,返回队首值并移除

 

l  队列为空时,会阻塞等待,如果在指定时间内队列还为空则返回 null

 

 

 

ArrayBlockingQueuepoll(time,unit)方法源码如下:

 

 

 

public E poll(long timeout, TimeUnit unitthrows InterruptedException {

 

        long nanos = unit.toNanos(timeout);

 

        final ReentrantLock lock = this.lock;

 

        lock.lockInterruptibly();  // 获取锁

 

        try {

 

            while (count == 0) {

 

                if (nanos <= 0L)

 

                    return null;

 

                nanos = notEmpty.awaitNanos(nanos); // 使线程等待指定的时间

 

            }

 

            return dequeue();  // 出队

 

        } finally {

 

            lock.unlock();  // 解锁

 

        }

 

    }

 

从上面代码可以看出,执行poll(time,unit)方法时,分为以下几个步骤:

 

 

 

l  先是要获取锁。

 

l  而后判断count是否等于0,如果等于0则证明队列为空,会阻塞等待;否则执行dequeue()方法做元素的出队。

 

l  最后解锁。

 

 

 

dequeue()方法此处不再赘述。

 

 

 

 

 

3.8.     remove()

 

执行remove()方法后有两种结果:

 

 

 

l  队列不为空时,返回队首值并移除

 

l  队列为空时,抛出异常

 

 

 

ArrayBlockingQueueremove()方法其实是调用了父类AbstractQueueremove()方法,源码如下:

 

 

 

public E remove() {

 

        E x = poll();

 

        if (x != null)

 

            return x;

 

        else

 

            throw new NoSuchElementException();

 

}

 

 

 

从上面代码可以看出,remove()直接调用了poll()方法。如果poll()方法返回结果为null,则抛出NoSuchElementException异常。

 

 

 

poll()方法此处不再赘述。

 

 

 

3.9.     peek()

 

执行peek()方法后有两种结果:

 

 

 

l  队列不为空时,返回队首值但不移除

 

l  队列为空时,返回null

 

 

 

 

 

peek()方法源码如下:

 

 

 

public E peek() {

 

        final ReentrantLock lock = this.lock;

 

        lock.lock(); // 加锁

 

        try {

 

            return itemAt(takeIndex); // 空则返回null

 

        } finally {

 

            lock.unlock();  // 解锁

 

        }

 

}

 

 

 

final E itemAt(int i) {

 

        return (E) items[i];

 

}

 

 

 

从上面代码可以看出,peek()方法比较简单,直接就是获取了数组里面的索引为takeIndex的元素。

 

 

 

3.10.            element()

 

执行element()方法后有两种结果:

 

 

 

l  队列不为空时,返回队首值但不移除

 

l  队列为空时,抛出异常

 

 

 

 

 

element()方法其实是调用了父类AbstractQueueelement()方法,源码如下:

 

 

 

public E element() {

 

        E x = peek();

 

        if (x != null)

 

            return x;

 

        else

 

            throw new NoSuchElementException();

 

}

 

 

 

从上面代码可以看出,执行element()方法时,先是获取peek()方法的结果,如果结果是null,则抛出NoSuchElementException异常。

 

 

 

 

 

4.   ArrayBlockingQueue的单元测试

 

 

 

ArrayBlockingQueue的单元测试如下:

 

 

 

 

 

package com.waylau.java.demo.datastructure;

 

 

 

import static org.junit.jupiter.api.Assertions.assertEquals;

 

import static org.junit.jupiter.api.Assertions.assertFalse;

 

import static org.junit.jupiter.api.Assertions.assertNotNull;

 

import static org.junit.jupiter.api.Assertions.assertNull;

 

import static org.junit.jupiter.api.Assertions.assertThrows;

 

import static org.junit.jupiter.api.Assertions.assertTrue;

 

 

 

import java.util.NoSuchElementException;

 

import java.util.concurrent.ArrayBlockingQueue;

 

import java.util.concurrent.BlockingQueue;

 

import java.util.concurrent.TimeUnit;

 

 

 

import org.junit.jupiter.api.Test;

 

 

 

/**

 

 * ArrayBlockingQueue Tests

 

 *

 

 * @since 1.0.0 202053

 

 * @author <a href="https://waylau.com">Way Lau</a>

 

 */

 

class ArrayBlockingQueueTests {

 

    @Test

 

    void testOffer() {

 

        // 初始化队列

 

        BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);

 

 

 

        // 测试队列未满时,返回 true

 

        boolean resultNotFull = queue.offer("Java");

 

        assertTrue(resultNotFull);

 

 

 

        // 测试队列满则,返回 false

 

        queue.offer("C");

 

        queue.offer("Python");

 

        boolean resultFull = queue.offer("C++");

 

        assertFalse(resultFull);

 

    }

 

 

 

    @Test

 

    void testPut() throws InterruptedException {

 

        // 初始化队列

 

        BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);

 

 

 

        // 测试队列未满时,直接插入没有返回值;

 

        queue.put("Java");

 

 

 

        // 测试队列满则, 会阻塞等待,一直等到队列未满时再插入。

 

        queue.put("C");

 

        queue.put("Python");

 

        queue.put("C++");  // 阻塞等待

 

    }

 

 

 

    @Test

 

    void testOfferTime() throws InterruptedException {

 

        // 初始化队列

 

        BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);

 

 

 

        // 测试队列未满时,返回 true

 

        boolean resultNotFull = queue.offer("Java", 5, TimeUnit.SECONDS);

 

        assertTrue(resultNotFull);

 

 

 

        // 测试队列满则,返回 false

 

        queue.offer("C");

 

        queue.offer("Python");

 

        boolean resultFull = queue.offer("C++", 5, TimeUnit.SECONDS); // 5

 

        assertFalse(resultFull);

 

    }

 

 

 

    @Test

 

    void testAdd() {

 

        // 初始化队列

 

        BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);

 

 

 

        // 测试队列未满时,返回 true

 

        boolean resultNotFull = queue.add("Java");

 

        assertTrue(resultNotFull);

 

 

 

        // 测试队列满则抛出异常

 

        queue.add("C");

 

        queue.add("Python");

 

 

 

        Throwable excpetion = assertThrows(IllegalStateException.class, () -> {

 

            queue.add("C++");// 抛异常

 

        });

 

 

 

        assertEquals("Queue full"excpetion.getMessage());

 

    }

 

 

 

    @Test

 

    void testPoll() throws InterruptedException {

 

        // 初始化队列

 

        BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);

 

 

 

        // 测试队列为空时,返回 null

 

        String resultEmpty = queue.poll();

 

        assertNull(resultEmpty);

 

 

 

        // 测试队列不为空时,返回队首值并移除

 

        queue.put("Java");

 

        queue.put("C");

 

        queue.put("Python");

 

        String resultNotEmpty = queue.poll();

 

        assertEquals("Java"resultNotEmpty);

 

    }

 

 

 

    @Test

 

    void testTake() throws InterruptedException {

 

        // 初始化队列

 

        BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);

 

 

 

        // 测试队列不为空时,返回队首值并移除

 

        queue.put("Java");

 

        queue.put("C");

 

        queue.put("Python");

 

        String resultNotEmpty = queue.take();

 

        assertEquals("Java"resultNotEmpty);

 

 

 

        // 测试队列为空时,会阻塞等待,一直等到队列不为空时再返回队首值

 

        queue.clear();

 

        String resultEmpty = queue.take(); // 阻塞等待

 

        assertNotNull(resultEmpty);

 

    }

 

 

 

    @Test

 

    void testPollTime() throws InterruptedException {

 

        // 初始化队列

 

        BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);

 

 

 

        // 测试队列不为空时,返回队首值并移除

 

        queue.put("Java");

 

        queue.put("C");

 

        queue.put("Python");

 

        String resultNotEmpty = queue.poll(5, TimeUnit.SECONDS);

 

        assertEquals("Java"resultNotEmpty);

 

 

 

        // 测试队列为空时,会阻塞等待,如果在指定时间内队列还为空则返回 null

 

        queue.clear();

 

        String resultEmpty = queue.poll(5, TimeUnit.SECONDS); // 等待5

 

        assertNull(resultEmpty);

 

    }

 

 

 

    @Test

 

    void testRemove() throws InterruptedException {

 

        // 初始化队列

 

        BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);

 

 

 

        // 测试队列为空时,抛出异常

 

        Throwable excpetion = assertThrows(NoSuchElementException.class, () -> {

 

            queue.remove();// 抛异常

 

        });

 

 

 

        assertEquals(nullexcpetion.getMessage());

 

 

 

        // 测试队列不为空时,返回队首值并移除

 

        queue.put("Java");

 

        queue.put("C");

 

        queue.put("Python");

 

        String resultNotEmpty = queue.remove();

 

        assertEquals("Java"resultNotEmpty);

 

    }

 

 

 

@Test

 

    void testPeek() throws InterruptedException {

 

        // 初始化队列

 

        Queue<String> queue = new ArrayBlockingQueue<String>(3);

 

 

 

        // 测试队列不为空时,返回队首值并但不移除

 

        queue.add("Java");

 

        queue.add("C");

 

        queue.add("Python");

 

        String resultNotEmpty = queue.peek();

 

        assertEquals("Java"resultNotEmpty);

 

        resultNotEmpty = queue.peek();

 

        assertEquals("Java"resultNotEmpty);

 

        resultNotEmpty = queue.peek();

 

        assertEquals("Java"resultNotEmpty);

 

 

 

        // 测试队列为空时,返回null

 

        queue.clear();

 

        String resultEmpty = queue.peek();

 

        assertNull(resultEmpty);

 

    }

 

 

 

    @Test

 

    void testElement() throws InterruptedException {

 

        // 初始化队列

 

        Queue<String> queue = new ArrayBlockingQueue<String>(3);

 

 

 

        // 测试队列不为空时,返回队首值并但不移除

 

        queue.add("Java");

 

        queue.add("C");

 

        queue.add("Python");

 

        String resultNotEmpty = queue.element();

 

        assertEquals("Java"resultNotEmpty);

 

        resultNotEmpty = queue.element();

 

        assertEquals("Java"resultNotEmpty);

 

        resultNotEmpty = queue.element();

 

        assertEquals("Java"resultNotEmpty);

 

 

 

        // 测试队列为空时,抛出异常

 

        queue.clear();

 

        Throwable excpetion = assertThrows(NoSuchElementException.class, () -> {

 

            queue.element();// 抛异常

 

        });

 

 

 

        assertEquals(nullexcpetion.getMessage());

 

    }

 

}

 

 

 

5.   ArrayBlockingQueue的应用案例

 

以下是一个生产者-消费者的示例。该示例模拟了1个生产者,2个消费者。当队列满时,则会阻塞生产者生产;当队列空时,则会阻塞消费者消费。

 

 

 

package com.waylau.java.demo.datastructure;

 

 

 

import java.util.concurrent.ArrayBlockingQueue;

 

import java.util.concurrent.BlockingQueue;

 

 

 

/**

 

 * ArrayBlockingQueue Demo

 

 *

 

 * @since 1.0.0 202053

 

 * @author <a href="https://waylau.com">Way Lau</a>

 

 */

 

public class ArrayBlockingQueueDemo {

 

 

 

    public static void main(String[] args) {

 

        BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);

 

       

 

        // 1个生产者

 

        Producer p = new Producer(queue);

 

       

 

        // 2个消费者

 

        Consumer c1 = new Consumer("c1"queue);

 

        Consumer c2 = new Consumer("c2"queue);

 

       

 

        // 启动线程

 

        new Thread(p).start();

 

        new Thread(c1).start();

 

        new Thread(c2).start();

 

    }

 

}

 

 

 

class Producer implements Runnable {

 

    private final BlockingQueue<String> queue;

 

 

 

    Producer(BlockingQueue<String> queue) {

 

        this.queue = queue;

 

    }

 

 

 

    public void run() {

 

        try {

 

            while (true) {

 

                // 模拟耗时操作

 

                Thread.sleep(1000L);

 

 

 

                queue.put(produce());

 

            }

 

        } catch (InterruptedException ex) {

 

            ex.printStackTrace();

 

        }

 

    }

 

 

 

    String produce() {

 

        String apple = "apple: " + System.currentTimeMillis();

 

        System.out.println("produce " + apple);

 

        return apple;

 

    }

 

}

 

 

 

class Consumer implements Runnable {

 

    private final BlockingQueue<String> queue;

 

 

 

    private final String name;

 

 

 

    Consumer(String name, BlockingQueue<String> queue) {

 

        this.queue = queue;

 

        this.name = name;

 

    }

 

 

 

    public void run() {

 

        try {

 

            while (true) {

 

                // 模拟耗时操作

 

                Thread.sleep(2000L);

 

 

 

                consume(queue.take());

 

            }

 

        } catch (InterruptedException ex) {

 

            ex.printStackTrace();

 

        }

 

    }

 

 

 

    void consume(Object x) {

 

        System.out.println(this.name + " consume " + x);

 

    }

 

}

 

 

 

 

 

 

 

 

 

运行上述程序,输出内容如下:

 

 

 

produce apple: 1590308383034

 

c2 consume apple: 1590308383034

 

produce apple: 1590308384034

 

c1 consume apple: 1590308384034

 

produce apple: 1590308385036

 

c2 consume apple: 1590308385036

 

produce apple: 1590308386036

 

c1 consume apple: 1590308386036

 

produce apple: 1590308387036

 

c2 consume apple: 1590308387036

 

produce apple: 1590308388036

 

c1 consume apple: 1590308388036

 

produce apple: 1590308389041

 

c2 consume apple: 1590308389041

 

produce apple: 1590308390041

 

c1 consume apple: 1590308390041

 

produce apple: 1590308391042

 

c2 consume apple: 1590308391042

 

produce apple: 1590308392042

 

c1 consume apple: 1590308392042

 

6.   参考引用

 

本系列归档至《Java数据结构及算法实战》:https://github.com/waylau/java-data-structures-and-algorithms-in-action
《数据结构和算法基础(Java语言实现)》(柳伟卫著,北京大学出版社出版):https://item.jd.com/13014179.html

展开阅读全文
加载中

作者的其它热门文章

打赏
0
0 收藏
分享
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部