Java使用阻塞队列BlockingQueue实现生产者消费者的方法_Java

什么是阻塞队列

阻塞队列(blockingqueue)是一个支持两个附加操作的队列。这两个附加的操作支持阻塞的插入和移除方法。

1、支持阻塞的插入方法：意思是当队列满时，队列会阻塞插入元素的线程，直到队列不满。
2、支持阻塞的移除方法：意思是在队列为空时，获取元素的线程会等待队列变为非空。

阻塞队列常用于生产者和消费者的场景，生产者是向队列里添加元素的线程，消费者是从队列里取元素的线程。阻塞队列就是生产者用来存放元素、消费者用来获取元素的容器。

java中提供了几个对blockingqueue的实现类，如: arrayblockingqueue, linkedblockingqueue, priorityblockingqueue, synchronousqueue 等

在处理生产者/消费者问题上我们将会使用arrayblockingqueue来实现，如下是我们需知道的重要方法：

put(e e): 这个方法用于向队列中插入元素，如果队列已满，需要等待可用的这间。
e take(): 这个方法用于从队列头部获取或者移除元素，如果队列为空则需要等待可用的元素。

使用blockingqueue来解决生产者/消费者示例

mantou类

producer产生的普通java对象，并添加到队列中。

/**
 * producer产生的馒头类
 * @author itmyhome
 *
 */
public class mantou {
    private string mantou;
    public mantou(string mantou) {
        this.mantou = mantou;
    }
    public string getmantou() {
        return mantou;
    }
    public void setmantou(string mantou) {
        this.mantou = mantou;
    }
}

producer生产者类

producer这个类会产生消息并将其放入队列中。

import java.util.concurrent.blockingqueue;
public class producer implements runnable {
    blockingqueue<mantou> queue;
    public producer(blockingqueue<mantou> queue) {
        this.queue = queue;
    }
    @override
    public void run() {
        // 生产馒头
        for (int i = 0; i < 100; i++) {
            mantou mt = new mantou("" + i);
            try {
                thread.sleep(100);
                queue.put(mt);
                system.out.println("生产馒头: " + mt.getmantou());
            } catch (interruptedexception e) {
                e.printstacktrace();
            }
        }
        // 添加退出消息
        mantou msg = new mantou("exit");
        try {
            queue.put(msg);
        } catch (interruptedexception e) {
            e.printstacktrace();
        }
    }
}

consumer消费者类

consumer类会从队列获取消息进行处理。如果获取的是退出消息则结束。

import java.util.concurrent.blockingqueue;
public class consumer implements runnable {
    blockingqueue<mantou> queue;
    public consumer(blockingqueue<mantou> queue) {
        this.queue = queue;
    }
    @override
    public void run() {
        try {
            mantou mantou;
            // 获取并处理消息直到接收到“exit”消息
            while (!(mantou = queue.take()).getmantou().equals("exit")) {
                thread.sleep(100);
                system.out.println("消费馒头： " + mantou.getmantou());
            }
        } catch (interruptedexception e) {
            e.printstacktrace();
        }
    }
}

producerconsumerservice

生产者/消费者的服务类将会产生固定大小的blockingqueue，生产者和消费者同时共享该blockingqueue，该服务类会起启动生产者和消费者线程。

import java.util.concurrent.arrayblockingqueue;
import java.util.concurrent.blockingqueue;
/**
 * @author itmyhome
 *
 */
public class producerconsumerservice {
    public static void main(string[] args) {
        // 创建大小为10的 blockingqueue
        blockingqueue<mantou> queue = new arrayblockingqueue<mantou>(10);
        producer producer = new producer(queue);
        consumer consumer = new consumer(queue);
        // 开启 producer线程向队列中生产消息
        new thread(producer).start();
        //开启 consumer线程 中队列中消费消息
        new thread(consumer).start();
        system.out.println("producer and consumer has been started");
    }
}

程序运行结果：

producer and consumer has been started
生产馒头： 0
生产馒头： 1
消费馒头： 0
消费馒头： 1
生产馒头： 2
消费馒头： 2
生产馒头： 3
消费馒头： 3
生产馒头： 4
消费馒头： 4
生产馒头： 5
消费馒头： 5
生产馒头： 6
消费馒头： 6
......