多线程 - 阻塞队列

Dear 丶 2024-03-22 16:23 83阅读 0赞

## 阻塞队列 ##

阻塞队列是一种特殊的队列，它支持在队列为空时等待从队列中获取元素的操作，或在队列已满时等待插入元素的操作。  
当队列为空时，从队列中获取元素的操作会被阻塞，直到队列中有可用元素；  
当队列已满时，插入元素的操作会被阻塞，直到队列中有空闲空间。  
Java 提供了一些内置的阻塞队列，比如 ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue 等。  
其中，ArrayBlockingQueue 是一种基于数组实现的阻塞队列，  
而 LinkedBlockingQueue 则是一种基于链表实现的阻塞队列，  
SynchronousQueue 则是一种特殊的阻塞队列，它的容量为 0，每个插入操作都必须等待一个相应的删除操作，反之亦然。

阻塞队列的主要优点是它可以帮助程序员简化多线程编程， 因为它可以避免显式地使用 wait() 和 notify() 等低级的线程同步原语，  
从而降低了编程的复杂度和错误的发生概率。另外，阻塞队列还可以用于实现一些高级的线程协作模式，比如生产者-消费者模式等。

## Java 中常见的阻塞队列有以下几种： ##

ArrayBlockingQueue：一个由数组支持的有界阻塞队列。底层实现使用了可重入锁 ReentrantLock 和两个条件变量 notEmpty 和 notFull 来保证线程安全和阻塞。
    
    LinkedBlockingQueue：一个由链表支持的可选有界阻塞队列。底层实现使用了可重入锁 ReentrantLock 和两个条件变量 notEmpty 和 notFull 来保证线程安全和阻塞。与 ArrayBlockingQueue 不同，它的容量可以选择无限大。
    
    PriorityBlockingQueue：一个支持优先级排序的无界阻塞队列。底层实现使用了可重入锁 ReentrantLock 和一个条件变量 available 来保证线程安全和阻塞。元素插入到队列中时会根据元素的优先级自动排序，优先级高的元素会被优先处理。
    
    SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等待一个相应的删除操作，反之亦然。底层实现使用了公平模式和非公平模式两种机制。
    
    DelayQueue：一个支持延迟删除元素的无界阻塞队列。元素只有在过了指定的延迟时间后才能从队列中取出。底层实现使用了可重入锁 ReentrantLock 和一个条件变量 available 来保证线程安全和阻塞。

除了上述几种阻塞队列外，Java 还提供了一些特殊的队列实现，比如 ConcurrentLinkedQueue、ConcurrentLinkedDeque、LinkedTransferQueue 等，它们各有特点，可以根据实际需求选择合适的队列实现。

## 常见的？ ##

Java 中常用的阻塞队列是 ArrayBlockingQueue 和 LinkedBlockingQueue，因为它们的实现比较简单，并且功能齐全，支持多线程并发操作。  
其中，ArrayBlockingQueue 是一个有界队列，可以指定队列的最大容量，而 LinkedBlockingQueue 是一个可选有界队列，容量可以选择无限大。  
对于需要使用优先级队列的场景，PriorityBlockingQueue 是一个不错的选择。

SynchronousQueue 和 DelayQueue 相对来说使用较少，比较特殊，需要根据具体情况来选择是否使用。

## BlockingQueue ##

import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.ArrayBlockingQueue;
    
    /**
     * 创建了一个大小为 5 的 ArrayBlockingQueue，并使用 put() 方法添加了五个整数。
     *
     * 当我们再次尝试添加第六个元素时，由于队列已经满了，offer() 方法会返回 false。
     * 然后我们使用 take() 方法获取并移除队列头部的元素，并使用 poll() 方法获取并移除另一个元素。
     * 我们还使用了 isEmpty()、remainingCapacity() 和 isFull() 等方法来判断队列是否为空、队列的剩余容量以及队列是否已满。
     *
     * 请注意，在使用 put() 或 take() 方法时，如果队列已经满了或者为空，
     * 那么线程将会被阻塞直到队列中有足够的空间或者元素可供消费。
     *
     * 而对于 offer() 和 poll() 方法，则不会阻塞线程，如果队列已满或者为空，它们会立即返回 false 或者 null。
     *
     */
    public class MyBlockQueueExample {
        
    
        public static void main(String[] args) throws InterruptedException {
        
    
            // 创建一个阻塞队列，大小为 5
            BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);
    
            // put() 方法向队列中添加元素，如果队列已满则会阻塞
            queue.put(1);
            queue.put(2);
            queue.put(3);
            queue.put(4);
            queue.put(5);
    
            // offer() 方法向队列中添加元素，如果队列已满则返回 false
            boolean isAdded = queue.offer(6);
            System.out.println("是否成功添加元素：" + isAdded);
    
            // take() 方法从队列头部获取并移除元素，如果队列为空则会阻塞
            int element = queue.take();
            System.out.println("获取的队列头部元素为：" + element);
    
            // poll() 方法从队列头部获取并移除元素，如果队列为空则返回 null
            Integer polledElement = queue.poll();
            System.out.println("获取并移除的队列头部元素为：" + polledElement);
    
            // isEmpty() 方法判断队列是否为空
            boolean isEmpty = queue.isEmpty();
            System.out.println("队列是否为空：" + isEmpty);
    
            // remainingCapacity() 方法获取队列的剩余容量
            int remainingCapacity = queue.remainingCapacity();
            System.out.println("队列的剩余容量为：" + remainingCapacity);
    
            // isFull() 方法判断队列是否已满
            boolean isFull = (queue.remainingCapacity() == 0);
            System.out.println("队列是否已满：" + isFull);
        }
    }

## ArrayBlockingQueue ##

import java.util.concurrent.ArrayBlockingQueue;
    
    public class MyArrayBlockingQueueExample {
        
    
        public static void main(String[] args) throws InterruptedException {
        
            
            // 创建一个阻塞队列，初始化容量为5
            ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(5);
            
            // 向队列里添加一个新的元素
            queue.put("A");
            queue.put("B");
            queue.put("C");
            queue.put("D");
            queue.put("E");
            
            // 因为队列已经满了，所以这里会阻塞
            // queue.put("F");
            
            // 检索然后移除第一个元素
            String firstElement = queue.take();
            System.out.println("Retrieved element: " + firstElement);
            
            // 向队列里添加一个新的元素
            queue.put("F");
            
            // 检索然后移除所有元素
            while (!queue.isEmpty()) {
        
                String element = queue.take();
                System.out.println("Retrieved element: " + element);
            }
        }
    }

## LinkedBlockingQueue ##

import java.util.concurrent.LinkedBlockingQueue;
    
    /**
     * 创建了一个容量为 3 的 LinkedBlockingQueue 对象，并添加了三个元素。
     * 然后我们从队列中取出第一个元素，并打印输出。
     *
     * 接下来我们查看了队列的大小，并判断了队列是否为空。
     * 最后我们使用 for 循环遍历了队列中的所有元素，并将其打印输出。
     */
    public class MyLinkedBlockingQueueExample {
        
        public static void main(String[] args) {
        
            LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<>(3);
    
            // 添加元素到队列
            try {
        
                queue.put("apple");
                queue.put("banana");
                queue.put("orange");
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
    
            // 从队列中取出元素
            String first = null;
            try {
        
                first = queue.take();
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
            System.out.println("First element: " + first);
    
            // 查看队列大小
            System.out.println("Queue size: " + queue.size());
    
            // 判断队列是否为空
            System.out.println("Queue is empty: " + queue.isEmpty());
    
            // 遍历队列中的元素
            System.out.println("Queue elements:");
            for (String element : queue) {
        
                System.out.println(element);
            }
        }
    }