【多线程】（三）阻塞式队列和生产者消费者模型

男娘i 2023-10-13 17:33 20阅读 0赞

### 文章目录 ###

*  一、什么是阻塞式队列
 *  二、Java库中的阻塞式队列
 *  三、阻塞式队列的模拟实现
 *  四、生产者消费者模型
 *   *  4.1 什么是生产者消费者模型
     *  4.2 基于阻塞队列实现简单的生产者消费者模型

--------------------

# 一、什么是阻塞式队列 #

**阻塞式队列（Blocking Queue）** 是一种特殊的队列，它提供了**线程安全的阻塞操作，以实现在队列为空或已满时的等待和阻塞**。

阻塞式队列通常用于**多线程环境下的生产者-消费者模型**，其中**生产者线程向队列中插入元素，消费者线程从队列中获取元素**。阻塞式队列提供了一种同步机制，使得**生产者线程在队列已满时等待插入，而消费者线程在队列为空时等待获取**。

**阻塞式队列的常用操作包括：**

1.  插入元素：
    
     *  `put(element)`: 在队列的尾部插入指定元素，如果队列已满，则阻塞等待空间可用。
2.  获取元素：
    
     *  `take()`: 从队列的头部获取并删除元素，如果队列为空，则阻塞等待元素可用。

阻塞式队列的**特点是提供了线程安全的阻塞操作，简化了多线程环境下的同步管理**。通过阻塞式队列，可以实现高效的生产者-消费者模型，并且避免了手动进行线程间的等待和通知操作。

# 二、Java库中的阻塞式队列 #

Java标准库中提供了多个阻塞式队列的实现，下面是几个常用的阻塞式队列实现及其说明：

<table> 
 <thead> 
  <tr> 
   <th align="left">阻塞式队列实现</th> 
   <th align="left">说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left"><code>ArrayBlockingQueue</code></td> 
   <td align="left"><strong>基于数组的有界阻塞队列</strong>。在创建时需要指定容量，插入操作会在队列已满时阻塞，获取操作会在队列为空时阻塞。</td> 
  </tr> 
  <tr> 
   <td align="left"><code>LinkedBlockingQueue</code></td> 
   <td align="left"><strong>基于链表的可选有界或无界阻塞队列</strong>。如果在创建时指定容量，则是有界队列；如果没有指定容量，则是无界队列。无界队列可以在没有上限的情况下存储元素。</td> 
  </tr> 
  <tr> 
   <td align="left"><code>PriorityBlockingQueue</code></td> 
   <td align="left"><strong>基于优先级的无界阻塞队列</strong>。元素按照指定的优先级顺序进行排序，每次获取操作会返回具有最高优先级的元素。</td> 
  </tr> 
  <tr> 
   <td align="left"><code>DelayQueue</code></td> 
   <td align="left"><strong>基于延迟时间的无界阻塞队列</strong>。队列中的元素必须实现 <code>Delayed</code> 接口，元素的延迟时间决定了元素可以从队列中获取的时间。</td> 
  </tr> 
  <tr> 
   <td align="left"><code>SynchronousQueue</code></td> 
   <td align="left"><strong>一个没有存储的阻塞队列</strong>。每个插入操作必须等待相应的删除操作，反之亦然。它主要用于线程之间的直接传输数据。</td> 
  </tr> 
 </tbody> 
</table>

这些阻塞式队列的实现都位于 `java.util.concurrent` 包下，都是线程安全的，适用于多线程环境下的并发编程。它们提供了不同的功能和特性，可以根据具体的需求选择合适的实现方式。通过使用这些阻塞式队列，可以简化多线程编程中的同步管理，并提高系统的性能和可靠性。

# 三、阻塞式队列的模拟实现 #

**实现代码如下：**

class MyBlockingQueue {
        
        // 尾入头出
        private int[] items = new int[1000]; // [head, tail)
        private int head = 0;
        private int tail = 0;
        private int size = 0;
    
        // 入队列
        public void put(int value) throws InterruptedException {
        
            synchronized (this) {
        
    
                // 当wait被唤醒了，此时的if条件可能还成立，因此将if改成while
                /*if (size == items.length) {
                    // 如果队列满了，要进行阻塞
                    this.wait();
                    // return;
                }*/
    
                while (size == items.length) {
        
                    // 如果队列满了，要进行阻塞
                    this.wait();
                }
    
    
                items[tail] = value;
                tail++;
    
                // 方式一：不直观
                // tail = tail % items.length;
    
                // 方式二：直观、效率比求余更高
                if (tail >= items.length) {
        
                    tail = 0;
                }
    
                size++;
                // 生产一个元素，唤醒出队操作
                this.notify();
            }
        }
    
        // 出队列
        public int take() throws InterruptedException {
        
            int result = 0;
            synchronized (this) {
        
                /*if (size == 0) {
                    // 队列为空，要进行阻塞
                    this.wait();
                    // return null;
                }*/
    
                while (size == 0) {
        
                    // 队列为空，要进行阻塞
                    this.wait();
                }
    
                result = items[head];
                head++;
                if (head >= items.length) {
        
                    head = 0;
                }
                size--;
                // 消耗一个元素，唤醒入队操作
                this.notify();
            }
            return result;
        }
    
    }

以上代码实现了一个简单的阻塞队列（MyBlockingQueue），代码使用一个长度为1000的整数数组（items）作为队列的存储容器，head和tail分别表示队列的头部和尾部的索引，size表示队列中当前元素的数量。

**入队操作（put方法）：**

*  使用synchronized关键字对方法进行同步，确保在多线程环境下的线程安全性。
 *  当队列已满时，使用while循环而不是if语句进行等待，以避免线程被意外唤醒后直接执行后续操作。
 *  当队列有空间时，将元素放入队列的尾部（tail索引位置），并更新tail的值。
 *  对tail的更新有两种方式，方式一是使用取模操作将tail限制在数组范围内，方式二是通过条件判断来重置tail的值。
 *  增加队列的元素数量（size++）并唤醒可能在等待中的出队操作。

**出队操作（take方法）：**

*  同样使用synchronized关键字对方法进行同步，确保在多线程环境下的线程安全性。
 *  当队列为空时，使用while循环而不是if语句进行等待，以避免线程被意外唤醒后直接执行后续操作。
 *  当队列非空时，取出队列头部（head索引位置）的元素，并更新head的值。
 *  对head的更新采用方式二，如果head超出了数组的范围（head >= items.length），则将head重置为0。
 *  减少队列的元素数量（size–）并唤醒可能在等待中的入队操作。

# 四、生产者消费者模型 #

## 4.1 什么是生产者消费者模型 ##

生产者-消费者模型是一种**常见的并发编程模型，用于协调多个线程之间的数据交换和任务处理**。在该模型中，有两类角色：生产者和消费者。

*  **生产者（Producer）**：生产者负责生成（生产）数据或任务，并将其放入共享的数据结构（例如队列）中。生产者可以是单个线程或多个并发线程。
 *  **消费者（Consumer）**：消费者负责从共享的数据结构中获取（消费）数据或任务，并进行相应的处理。消费者可以是单个线程或多个并发线程。

生产者-消费者模型的**主要目标是解耦生产者和消费者，使它们能够独立运行，并以不同的速度进行操作**。通过引入共享的数据结构，可以在生产者和消费者之间建立一种同步机制，以确保线程安全和正确的数据交换顺序。

在实际应用中，生产者-消费者模型具有广泛的应用场景，例如**多线程任务处理、消息队列、线程池**等。它能够有效地利用系统资源，提高系统的并发性和响应性。

## 4.2 基于阻塞队列实现简单的生产者消费者模型 ##

public class ThreadDemo {
        
        public static void main(String[] args) {
        
            BlockingDeque<Integer> blockingDeque = new LinkedBlockingDeque<>();
    
            // 创建两个线程，t1作为生产者，t2作为消费者
            Thread t1 = new Thread(() -> {
        
                int count = 0;
                while (true){
        
                    try {
        
                        System.out.println("线程t1生产元素：" + count);
                        blockingDeque.put(count);
                        count++;
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
            });
    
            Thread t2 = new Thread(() -> {
        
                while (true){
        
                    Integer res = null;
                    try {
        
                        res = blockingDeque.take();
                        // Thread.sleep(500);
                        System.out.println("线程t2消费元素：" + res);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
            });
    
            t1.start();
            t2.start();
        }
    }

这段代码是一个简单的示例，演示了使用阻塞队列实现的生产者-消费者模型。

代码中创建了一个`BlockingDeque<Integer>`类型的阻塞队列`blockingDeque`，使用`LinkedBlockingDeque`作为具体的实现。这个阻塞队列可以存储整数类型的元素，并且支持阻塞的插入和移除操作。

然后，创建了两个线程：`t1`和`t2`，分别作为生产者和消费者。

*  生产者线程（`t1`）使用一个无限循环来模拟持续的生产操作。在每次循环中，它先打印当前生产的元素值，然后使用`blockingDeque.put(count)`将元素放入阻塞队列中。`blockingDeque.put(count)`是一个阻塞操作，如果队列已满，生产者线程会被阻塞，直到队列有空间可用。之后，生产者线程将计数值递增，并通过`Thread.sleep(500)`休眠一段时间，模拟生产的间隔。
 *  消费者线程（`t2`）同样使用一个无限循环来模拟持续的消费操作。在每次循环中，它使用`blockingDeque.take()`从阻塞队列中取出一个元素。`blockingDeque.take()`也是一个阻塞操作，如果队列为空，消费者线程会被阻塞，直到队列有元素可取。取出的元素被赋值给`res`变量，并打印出消费的元素值。然后，消费者线程继续下一轮循环。

这段代码演示了使用阻塞队列作为生产者和消费者之间的缓冲区，通过阻塞队列的阻塞和唤醒机制，实现了线程间的同步和数据交换。生产者线程不需要关心消费者线程的状态，消费者线程也不需要关心生产者线程的状态，它们通过共享的阻塞队列实现了解耦和线程安全。