Java并发编程 - 阻塞队列BlockingQueue

在Java并发编程中，BlockingQueue是一个非常有用的工具类，它提供了一种线程安全的方式来管理对象的队列。BlockingQueue属于java.util.concurrent包的一部分，它的特点是当队列为空时，从队列中获取元素的操作会阻塞，直到队列中出现新的元素；同样，当队列满时，往队列中添加元素的操作会阻塞，直到队列中有可用的空间。

BlockingQueue接口

BlockingQueue<E>是一个接口，它扩展了Queue<E>接口，并且添加了一些阻塞方法。这些方法包括：

• void put(E e)：将一个元素插入到队列中，如果队列满，则阻塞直到队列中有可用空间。
• E take()：从队列中移除并返回一个元素，如果队列为空，则阻塞直到队列中有元素可用。
• boolean offer(E e)：尝试将一个元素插入到队列中，如果成功则返回true，否则返回false。
• boolean offer(E e, long timeout, TimeUnit unit)：尝试将一个元素插入到队列中，如果队列满，则等待指定的时间，如果成功插入则返回true，否则返回false。
• E poll(long timeout, TimeUnit unit)：尝试从队列中移除并返回一个元素，如果队列为空，则等待指定的时间，如果成功移除则返回该元素，否则返回null。

常见的BlockingQueue实现

BlockingQueue有几个常用的实现类：

1. ArrayBlockingQueue：

   • 基于数组的阻塞队列。
   • 固定容量。
   • 支持公平和非公平两种策略。

2. LinkedBlockingQueue：

   • 基于链表的阻塞队列。
   • 可以设置固定容量，也可以是无界的（默认情况下为无界）。

3. PriorityBlockingQueue：

   • 基于优先级堆的阻塞队列。
   • 无界，但需要实现Comparable接口或提供一个Comparator来确定元素的优先级。

4. SynchronousQueue：

   • 特殊的阻塞队列，不存储元素，而是直接在生产者和消费者之间传递元素。
   • 非常适合用于构建生产者-消费者模型。

5. DelayQueue：

   • 用于存储延时元素的阻塞队列。
   • 队列中的元素会在指定的时间之后变得可用。

示例代码

下面是一个简单的示例，展示了如何使用LinkedBlockingQueue作为阻塞队列：

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public class BlockingQueueExample {

    private final LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);

    public void produce() {
        for (int i = 0; i < 100; i++) {
            try {
                queue.put(i); // 生产数据
                System.out.println("Produced: " + i);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    public void consume() {
        while (true) {
            try {
                Integer value = queue.poll(1, TimeUnit.SECONDS); // 消费数据
                if (value != null) {
                    System.out.println("Consumed: " + value);
                } else {
                    System.out.println("No more elements to consume.");
                    break;
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        BlockingQueueExample example = new BlockingQueueExample();

        Thread producer = new Thread(example::produce);
        Thread consumer = new Thread(example::consume);

        producer.start();
        consumer.start();

        producer.join();
        consumer.join();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

在这个示例中，我们创建了一个LinkedBlockingQueue实例，并设置了最大容量为10。生产者线程通过put方法将元素加入队列，而消费者线程通过poll方法从队列中取出元素。当队列满时，put方法会阻塞，直到队列中有可用的空间；当队列为空时，poll方法会阻塞，直到队列中有元素可用。

使用场景

BlockingQueue非常适合以下场景：

1. 生产者-消费者模式：在多线程环境中，生产者负责生成数据，消费者负责处理数据，中间通过BlockingQueue传递数据。
2. 任务队列：将任务放入队列中，由工作线程异步处理。
3. 缓存管理：用于实现有限大小的缓存，当缓存满时，新的元素会等待旧的元素被移除。

总结

BlockingQueue是一个非常强大的工具，它简化了多线程编程中的同步问题。通过使用阻塞队列，可以避免显式的同步代码，使程序更简洁、更易于理解。选择合适的BlockingQueue实现取决于具体的应用场景和需求。