Java 通过阻塞队列实现生产者消费者模式

更多 Java 并发编程方面的文章,请参见文集《Java 并发编程》


阻塞队列 Blocking Queue

  • 当队列空时,获取元素的线程会等待
  • 当队列满时,存储元素的线程会等待

提供的方法:

  • 插入元素:

    • add(e):抛出异常
    • offer(e):返回特殊值
    • put(e):一直阻塞
    • offer(e,time,unit):超时退出
  • 移除元素:

    • remove():抛出异常
    • poll():返回特殊值
    • take():一直阻塞
    • poll(time,unit):超时退出

JDK 7 提供了 7 个阻塞队列

  • ArrayBlockingQueue :一个由数组结构组成的 有界 阻塞队列。
    • 此队列按照先进先出(FIFO)的原则对元素进行排序。
    • 默认情况下不保证访问者公平的访问队列,所谓公平访问队列是指阻塞的所有生产者线程或消费者线程,当队列可用时,可以按照阻塞的先后顺序访问队列,即先阻塞的生产者线程,可以先往队列里插入元素,先阻塞的消费者线程,可以先从队列里获取元素。通常情况下为了保证公平性会降低吞吐量。我们可以使用以下代码创建一个公平的阻塞队列:
      ArrayBlockingQueue fairQueue = new ArrayBlockingQueue(1000,true);
  • LinkedBlockingQueue :一个由链表结构组成的 有界 阻塞队列。
    • 此队列按照先进先出(FIFO)的原则对元素进行排序。
  • PriorityBlockingQueue :一个支持优先级排序的 无界 阻塞队列。
    • 默认情况下元素采取自然顺序排列,也可以通过比较器 comparator 来指定元素的排序规则。元素按照升序排列。
  • DelayQueue:一个使用优先级队列实现的无界阻塞队列。
  • SynchronousQueue:一个不存储元素的阻塞队列。
  • LinkedTransferQueue:一个由链表结构组成的无界阻塞队列。
  • LinkedBlockingDeque:一个由链表结构组成的双向阻塞队列。

关于 通过 wait 和 notify 实现生产者消费者模式,可以参考 链接
关于 通过 Lock 和 竞争条件 Condition 实现生产者消费者模式,可以参考 链接

利用阻塞队列实现生产者消费者模式,代码如下:

public class BlockingQueue_Test {
    private static final int MAX_CAPACITY = 10;
    private static ArrayBlockingQueue<Object> goods = new ArrayBlockingQueue<Object>(MAX_CAPACITY);

    public static void main(String[] args) {
        (new ProducerThread()).start();

        (new ConsumerThread()).start();
    }

    static class ProducerThread extends Thread {
        public void run() {
            while (true) {
                // 每隔 1000 毫秒生产一个商品
                try {
                    Thread.sleep(1000);

                    goods.put(new Object());
                    System.out.println("Produce goods, total: " + goods.size());
                } catch (InterruptedException e) {
                }
            }
        }
    }

    static class ConsumerThread extends Thread {
        public void run() {
            while (true) {
                // 每隔 500 毫秒消费一个商品
                try {
                    Thread.sleep(500);

                    goods.take();
                    System.out.println("Consume goods, total: " + goods.size());
                } catch (InterruptedException e) {
                }
            }
        }
    }
}

阻塞队列的实现原理

ArrayBlockingQueue 为例,实际上使用了 ReentrantLock 和 Condition。

构造方法:

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();
}

插入元素,如果队列已满,则阻塞 notFull.await();:

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();
    }
}

移除元素,如果队列已空,则阻塞 notEmpty.await();:

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}

引用:
聊聊并发(七)——Java中的阻塞队列

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容

  • 相关文章Java并发编程(一)线程定义、状态和属性 Java并发编程(二)同步Java并发编程(三)volatil...
    刘望舒阅读 5,251评论 1 31
  • 阻塞队列(BlockingQueue)是一个支持两个附加操作的队列。这两个附加的操作是:在队列为空时,获取元素的线...
    端木轩阅读 1,015评论 0 2
  • 队列 队列是先进先出(FIFO)的线性表。在具体应用中通常用链表或者数组来实现。队列只允许在后端(称为rear)进...
    Showdy阅读 429评论 0 0
  • 1.阻塞队列定义阻塞队列常用于生产者和消费者的场景,生产者是往队列里添加元素的线程,消费者是从队列里拿元素的线程。...
    SDY_0656阅读 440评论 0 1
  • 亲爱的大木子: 你好吗?当你收到这封信的时候应该是2026年了,这是一封穿越时空的信件,不知道你收到这封信的心情会...
    木子不爱糖阅读 311评论 1 4