Java多线程：BlockingQueue实现生产者消费者模型

Posted 2022-06-20 java-starter

BlockingQueue

BlockingQueue、解决了多线程中，如何高效安全“传输”数据的问题。程序员无需关心什么时候阻塞线程，什么时候唤醒线程，该唤醒哪个线程。

方法介绍

BlockingQueue是Queue的子类

void put(E e)

插入指定元素，当BlockingQueue为满，则线程阻塞，进入Waiting状态，直到BlockingQueue有空闲空间再继续。
这里以ArrayBlockingQueue为例进行分析

void take()

队首出队，当BlockingQueue为空，则线程阻塞，进入Waiting状态，直到BlockingQueue不为空再继续。

int drainTo(Collection<? super E> c)

从队列中批量取出数据，并放入到另一个集合中，返回转移数据的数量，只需一次加锁和解锁。

BlockingQueue的实现类

ArrayBlockingQueue

    /*
     * Concurrency control uses the classic two-condition algorithm
     * found in any textbook.
     */

    /** Main lock guarding all access */
    final ReentrantLock lock;

    /** Condition for waiting takes */
    private final Condition notEmpty;

    /** Condition for waiting puts */
    private final Condition notFull;

基于数组实现的BlockingQueue，需要指定队列容量，可以指定是否为公平锁；只有一个ReentrantLock，生产者和消费者不能异步执行。

LinkedBlockingQueue

    /** Lock held by take, poll, etc */
    private final ReentrantLock takeLock = new ReentrantLock();

    /** Wait queue for waiting takes */
    private final Condition notEmpty = takeLock.newCondition();

    /** Lock held by put, offer, etc */
    private final ReentrantLock putLock = new ReentrantLock();

    /** Wait queue for waiting puts */
    private final Condition notFull = putLock.newCondition();

基于链表实现的BlockingQueue，可以指定队列容量，不指定队列容量默认为Integer.MAX_VALUE；有两个ReentrantLock，生产者和消费者可以异步执行。

BlockingQueue实现生产者消费者模型

• 缓冲区可以存放大量数据
• 生产者和消费者速度各不相同

public class MyThread42 
    public static void main(String[] args)
    
        final BlockingQueue<String> bq = new ArrayBlockingQueue<String>(10);
        Runnable producerRunnable = new Runnable()
        
            int i = 0;
            public void run()
            
                while (true)
                
                    try
                    
                        System.out.println("我生产了一个" + i++);
                        bq.put(i + "");
                        Thread.sleep(1000);
                    
                    catch (InterruptedException e)
                    
                        e.printStackTrace();
                    
                
            
        ;

        Runnable customerRunnable = new Runnable()
        
            public void run()
            
                while (true)
                
                    try
                    
                        System.out.println("我消费了一个" + bq.take());
                        Thread.sleep(3000);
                    
                    catch (InterruptedException e)
                    
                        e.printStackTrace();
                    
                
            
        ;
        Thread producerThread = new Thread(producerRunnable);
        Thread customerThread = new Thread(customerRunnable);
        producerThread.start();
        customerThread.start();
    

输出结果如下

我生产了一个0
我消费了一个1
我生产了一个1
我生产了一个2
我消费了一个2
我生产了一个3
我生产了一个4
我生产了一个5
我消费了一个3
我生产了一个6
我生产了一个7
我生产了一个8
我消费了一个4
我生产了一个9
我生产了一个10
我生产了一个11
我消费了一个5
我生产了一个12
我生产了一个13
我生产了一个14
我消费了一个6
我生产了一个15
我生产了一个16
我消费了一个7
我生产了一个17
我消费了一个8
我生产了一个18
我消费了一个9
我生产了一个19
我消费了一个10
我生产了一个20
我消费了一个11
我生产了一个21
我消费了一个12
我生产了一个22
我消费了一个13
我生产了一个23
我消费了一个14
我生产了一个24

······

生产者没有生产到BlockingQueue的容量（极限是10）之前，生产3个，消费1个，再生产到BlockingQueue的容量之后，生产一个消费一个，因为不能超过BlockingQueue的容量。