Java BlockingQueue 示例

今天我们将研究 Java BlockingQueue。`java.util.concurrent.BlockingQueue` 是一个 Java 队列，支持在检索和移除元素时等待队列变为非空，以及在添加元素时等待队列中有空间可用。

Java BlockingQueue

Java BlockingQueue 不接受 `null` 值，如果您尝试将 null 值存储在队列中，则会抛出 `NullPointerException`。Java BlockingQueue 实现是 线程安全 的。所有排队方法在性质上都是原子的，并使用内部锁或其他形式的并发控制。Java BlockingQueue 接口是 Java 集合框架的一部分，主要用于实现生产者消费者问题。我们不需要担心为生产者等待空间可用或为消费者等待对象可用，因为这些都由 BlockingQueue 的实现类处理。Java 提供了几种 BlockingQueue 实现，例如 `ArrayBlockingQueue`、`LinkedBlockingQueue`、`PriorityBlockingQueue`、`SynchronousQueue` 等。在实现 BlockingQueue 的生产者消费者问题时，我们将使用 ArrayBlockingQueue 实现。以下是您应该了解的一些重要方法。

• put(E e): 此方法用于向队列中插入元素。如果队列已满，则等待空间可用。
• E take(): This method retrieves and remove the element from the head of the queue. If queue is empty it waits for the element to be available.

现在让我们使用 Java BlockingQueue 实现生产者消费者问题。

Java BlockingQueue 示例 – 消息

这只是一个普通的 Java 对象，由生产者生成并添加到队列中。您也可以称之为负载或队列消息。

package com.journaldev.concurrency;

public class Message {
    private String msg;
    
    public Message(String str){
        this.msg=str;
    }

    public String getMsg() {
        return msg;
    }

}

Java BlockingQueue 示例 – 生产者

将创建消息并将其放入队列中的生产者类。

package com.journaldev.concurrency;

import java.util.concurrent.BlockingQueue;

public class Producer implements Runnable {

    private BlockingQueue queue;
    
    public Producer(BlockingQueue q){
        this.queue=q;
    }
    @Override
    public void run() {
        // 生成消息
        for(int i=0; i<100; i++){
            Message msg = new Message(""+i);
            try {
                Thread.sleep(i);
                queue.put(msg);
                System.out.println("Produced "+msg.getMsg());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        // 添加退出消息
        Message msg = new Message("exit");
        try {
            queue.put(msg);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

Java BlockingQueue 示例 – 消费者

将从队列中处理消息并在接收到退出消息时终止的消费者类。

package com.journaldev.concurrency;

import java.util.concurrent.BlockingQueue;

public class Consumer implements Runnable{

private BlockingQueue queue;
    
    public Consumer(BlockingQueue q){
        this.queue=q;
    }

    @Override
    public void run() {
        try{
            Message msg;
            // 在接收到退出消息之前消耗消息
            while((msg = queue.take()).getMsg() !="exit"){
            Thread.sleep(10);
            System.out.println("Consumed "+msg.getMsg());
            }
        }catch(InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Java BlockingQueue 示例 – 服务

最后，我们需要为生产者和消费者创建BlockingQueue服务。这个生产者消费者服务将创建具有固定大小的BlockingQueue，并与生产者和消费者共享。该服务将启动生产者和消费者线程，然后退出。

package com.journaldev.concurrency;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class ProducerConsumerService {

    public static void main(String[] args) {
        //创建大小为10的BlockingQueue
        BlockingQueue queue = new ArrayBlockingQueue<>(10);
        Producer producer = new Producer(queue);
        Consumer consumer = new Consumer(queue);
        //启动生产者将消息放入队列
        new Thread(producer).start();
        //启动消费者从队列中消费消息
        new Thread(consumer).start();
        System.out.println("Producer and Consumer has been started");
    }

}

上述Java BlockingQueue示例程序的输出如下所示。

Producer and Consumer has been started
Produced 0
Produced 1
Produced 2
Produced 3
Produced 4
Consumed 0
Produced 5
Consumed 1
Produced 6
Produced 7
Consumed 2
Produced 8
...

Java线程休眠用于在生产者和消费者之间以一定延迟生成和消费消息。