Java 阻塞队列实现生产者和消费者场景

太过爱你忘了你带给我的痛 2023-07-08 09:29 108阅读 0赞

### 1. 阻塞队列 ###

阻塞队列（BlockingQueue）是一个支持两个附加操作的队列。这两个附加的操作是：

*  在队列为空时，获取元素的线程会等待队列变为非空。
 *  当队列满时，存储元素的线程会等待队列可用。

当阻塞队列满时，如果生产者线程使用put方法向队列中添加元素，队列会一直阻塞生产者线程，直到拿到数据，或者响应中断退出。当队列空时，消费者使用take方法从队列中获取元素，队列也会阻塞消费者线程，直到队列可用。

### 2. 阻塞队列实现生产者和消费者实例 ###

生产者线程，ProducerThread.java:

import java.util.Random;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 生产者线程
     * @author syrdbt
     * @date 2020-02-26
     */
    public class ProducerThread implements Runnable{
        
    
        private LinkedBlockingQueue blockingQueue;
    
        public ProducerThread(LinkedBlockingQueue blockingQueue) {
        
            this.blockingQueue = blockingQueue;
        }
    
        @Override
        public void run() {
        
            Random random = new Random();
            while (true) {
        
                try {
        
                    this.produce(random.nextInt(5));
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
        }
    
        private void produce(int productNumber) throws InterruptedException {
        
            TimeUnit.SECONDS.sleep(productNumber);
            blockingQueue.put(productNumber);
            System.out.println("【生产者】生产了" + productNumber + "号产品，花费了" + productNumber + "秒，成功的添加"
                    + productNumber + "产品到了队列中");
        }
    }

消费者线程，ConsumerThread.java：

/**
     * 消费者线程
     * @author syrdbt
     * @date 2020-02-26
     */
    public class ConsumerThread implements Runnable{
        
    
        private LinkedBlockingQueue blockingQueue;
    
        public ConsumerThread(LinkedBlockingQueue blockingQueue) {
        
            this.blockingQueue = blockingQueue;
        }
    
        @Override
        public void run() {
        
            Random random = new Random();
            while (true) {
        
                try {
        
                    this.consume();
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
        }
    
        private void consume() throws InterruptedException {
        
            int productNumber = (int) blockingQueue.take();
            TimeUnit.SECONDS.sleep(productNumber);
            System.out.println("【消费者】消费了" + productNumber + "号产品，花费了" +
                    productNumber + "秒，成功的从队列中取出" + productNumber + "号产品");
        }
    }

测试程序，Main.java：

import java.util.concurrent.LinkedBlockingQueue;
    
    /**
     * @author syrdbt
     * @date 2020-02-26
     */
    public class Main {
        
        public static void main(String[] args) {
        
            LinkedBlockingQueue blockingQueue = new LinkedBlockingQueue();
            Thread producerThread = new Thread(new ProducerThread(blockingQueue));
            Thread consumerThread = new Thread(new ConsumerThread(blockingQueue));
            producerThread.start();
            consumerThread.start();
        }
    }

测试程序运行截图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4NzM3OTky_size_16_color_FFFFFF_t_70]

### 参考文献 ###

*  [聊聊并发（七）——Java中的阻塞队列][Java]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4NzM3OTky_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200226120746577.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4NzM3OTky,size_16,color_FFFFFF,t_70
[Java]: http://ifeve.com/java-blocking-queue/